/*-------------------------------------------------------------------------
 *
 * walsender.c
 *
 * WAL 发送器进程（walsender）是 Postgres 9.0 以来的新功能。
 * 它负责将 XLOG 从主服务器发送到单个接收者。
 * （请注意，可以同时存在多个 walsender 进程。）
 * 当备用服务器的 walreceiver 连接到主服务器并请求
 * XLOG 流复制时，postmaster 启动它。
 *
 * walsender 类似于常规后端，即连接和 walsender 进程之间
 * 具有一一对应关系，但它不是处理 SQL 查询，而是理解一小
 * 组特殊的复制模式命令。START_REPLICATION 命令开始
 * 将 WAL 流式传输到客户端。在流传输期间，walsender 会
 * 持续从磁盘读取 XLOG 记录并通过 COPY 协议将其发送到
 * 备用服务器，直到任一方通过退出 COPY 模式结束复制
 * （或直到连接关闭）。
 *
 * 正常终止是通过 SIGTERM，这指示 walsender
 * 关闭连接并在下一个方便的时刻退出(0)。紧急
 * 终止通过 SIGQUIT；与任何后端一样，walsender
 * 将在 SIGQUIT 时简单地中止并退出。关闭连接
 * 和 FATAL 错误被视为不仅仅是崩溃，而是大约正常的终止；
 * walsender 会快速退出而不发送更多的 XLOG 记录。
 *
 * 如果服务器关闭，检查点将在所有常规后端退出后发送
 * PROCSIG_WALSND_INIT_STOPPING。如果后端空闲或运行 SQL 查询 
 * 则会导致后端关闭，如果逻辑复制正在进行，则所有现有的
 * WAL 记录都会被处理，然后再关闭。否则，这将导致
 * walsender 切换到“停止”状态。在此状态下，walsender 将拒绝
 * 任何进一步的复制命令。检查点在所有 walsender 被确认
 * 停止后开始关闭检查点。当关闭检查点结束时，postmaster
 * 向我们发送 SIGUSR2。这指示 walsender 发送任何未处理的 WAL，
 * 包括关闭检查点记录，等待它复制到备用，然后退出。
 *
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/walsender.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <signal.h>
#include <unistd.h>

#include "access/printtup.h"
#include "access/timeline.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "access/xlogreader.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
#include "backup/basebackup.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_type.h"
#include "commands/dbcommands.h"
#include "commands/defrem.h"
#include "funcapi.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "miscadmin.h"
#include "nodes/replnodes.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/slot.h"
#include "replication/snapbuild.h"
#include "replication/syncrep.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "replication/walsender_private.h"
#include "storage/condition_variable.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "tcop/dest.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/portal.h"
#include "utils/ps_status.h"
#include "utils/timeout.h"
#include "utils/timestamp.h"

/*
 * WAL 数据消息中的最大数据有效负载。必须 >= XLOG_BLCKSZ。
 *
 * 我们对什么是合适的值没有一个很好的估计；在 walsender 和 walreceiver 
 * 中，每条消息都有一些开销，但另一方面发送大批量会降低 walsender 
 * 对信号的响应，因为信号仅在消息之间检查。128kB（默认 8k 块）目前似乎是一个合理的猜测。
 */
#define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)

/* 共享内存中的 WalSnds 数组 */
WalSndCtlData *WalSndCtl = NULL;

/* 我在共享内存数组中的插槽 */
WalSnd	   *MyWalSnd = NULL;

/* 全局状态 */
bool		am_walsender = false;	/* 我是一个 walsender 进程吗？ */
bool		am_cascading_walsender = false; /* 我是否将WAL级联到另一个备用？ */
bool		am_db_walsender = false;	/* 是否连接到数据库？ */

/* 用户可设置的walsender参数 */
int			max_wal_senders = 0;	/* 并发walsenders的最大数量 */
int			wal_sender_timeout = 60 * 1000; /* 发送一条WAL数据消息的最大时间 */
bool		log_replication_commands = false;

/*
 * WalSndWakeupRequest的状态
 */
bool		wake_wal_senders = false;

/*
 * 用于复制的xlogreader。请注意，执行物理复制的WAL发送器不需要xlogreader来读取WAL，但需要它来保持其工作的状态。
 */
static XLogReaderState *xlogreader = NULL;

/*
 * 这些变量跟踪我们当前发送的时间线状态。sendTimeLine标识时间线。如果sendTimeLineIsHistoric，时间线不是此服务器上的最新时间线，并且服务器的历史在sendTimeLineValidUpto处分叉。
 */
static TimeLineID sendTimeLine = 0;
static TimeLineID sendTimeLineNextTLI = 0;
static bool sendTimeLineIsHistoric = false;
static XLogRecPtr sendTimeLineValidUpto = InvalidXLogRecPtr;

/*
 * 我们已经发送了多远的WAL？这也在MyWalSnd->sentPtr中进行了宣传。（实际上，这是下一个要发送的WAL位置。）
 */
static XLogRecPtr sentPtr = InvalidXLogRecPtr;

/* 用于构建传出消息和处理回复消息的缓冲区。 */
static StringInfoData output_message;
static StringInfoData reply_message;
static StringInfoData tmpbuf;

/* 上一次ProcessRepliesIfAny()的时间戳。 */
static TimestampTz last_processing = 0;

/*
 * 上一次ProcessRepliesIfAny()看到从备用的回复的时间戳。如果wal_sender_timeout不需要激活，则设置为0。
 */
static TimestampTz last_reply_timestamp = 0;

/* 自上次回复以来，我们是否发送了请求回复的心跳消息？ */
static bool waiting_for_ping_response = false;

/*
 * 在以Copy模式流式传输WAL时，streamingDoneSending在我们发送了CopyDone后设置为true。之后我们不应该再发送任何CopyData消息。streamingDoneReceiving在我们从另一端收到CopyDone时设置为true。当两者都变为true时，是时候退出Copy模式了。
 */
static bool streamingDoneSending;
static bool streamingDoneReceiving;

/* 我们到了吗？ */
static bool WalSndCaughtUp = false;

/* 信号处理程序设置的标志，以便在主循环中稍后服务 */
static volatile sig_atomic_t got_SIGUSR2 = false;
static volatile sig_atomic_t got_STOPPING = false;

/*
 * 这在我们流式传输时设置。当未设置时，PROCSIG_WALSND_INIT_STOPPING信号将像SIGTERM那样处理。当设置时，主循环负责检查got_STOPPING并在设置时终止（在流式传输任何剩余的WAL后）。
 */
static volatile sig_atomic_t replication_active = false;

static LogicalDecodingContext *logical_decoding_ctx = NULL;
#ifdef FDD
static bool fc_FDB_Wal_SndKeepalive(void);
#endif

/* 一个将WAL位置与其写入时间关联的示例。 */
typedef struct
{
	XLogRecPtr	lsn;
	TimestampTz time;
} WalTimeSample;

/* 我们的时间样本缓冲区的大小。 */
#define LAG_TRACKER_BUFFER_SIZE 8192

/* 跟踪复制延迟的机制。 */
typedef struct
{
	XLogRecPtr	last_lsn;
	WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE];
	int			write_head;
	int			read_heads[NUM_SYNC_REP_WAIT_MODE];
	WalTimeSample last_read[NUM_SYNC_REP_WAIT_MODE];
} LagTracker;

static LagTracker *lag_tracker;

/* 信号处理程序 */
static void fc_WalSndLastCycleHandler(SIGNAL_ARGS);

/* 私有函数的原型 */
typedef void (*WalSndSendDataCallback) (void);
static void fc_WalSndLoop(WalSndSendDataCallback fc_send_data);
static void fc_InitWalSenderSlot(void);
static void fc_WalSndKill(int fc_code, Datum fc_arg);
static void fc_WalSndShutdown(void) pg_attribute_noreturn();
static void fc_XLogSendPhysical(void);
static void fc_XLogSendLogical(void);
static void fc_WalSndDone(WalSndSendDataCallback fc_send_data);
static XLogRecPtr fc_GetStandbyFlushRecPtr(TimeLineID *fc_tli);
static void fc_IdentifySystem(void);
static void fc_ReadReplicationSlot(ReadReplicationSlotCmd *fc_cmd);
static void fc_CreateReplicationSlot(CreateReplicationSlotCmd *fc_cmd);
static void fc_DropReplicationSlot(DropReplicationSlotCmd *fc_cmd);
static void fc_StartReplication(StartReplicationCmd *fc_cmd);
static void fc_StartLogicalReplication(StartReplicationCmd *fc_cmd);
static void fc_ProcessStandbyMessage(void);
static void fc_ProcessStandbyReplyMessage(void);
static void fc_ProcessStandbyHSFeedbackMessage(void);
static void fc_ProcessRepliesIfAny(void);
static void fc_ProcessPendingWrites(void);
static void fc_WalSndKeepalive(bool fc_requestReply, XLogRecPtr fc_writePtr);
static void fc_WalSndKeepaliveIfNecessary(void);
static void fc_WalSndCheckTimeOut(void);
static long fc_WalSndComputeSleeptime(TimestampTz fc_now);
static void fc_WalSndWait(uint32 fc_socket_events, long fc_timeout, uint32 fc_wait_event);
static void fc_WalSndPrepareWrite(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid, bool fc_last_write);
static void fc_WalSndWriteData(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid, bool fc_last_write);
static void fc_WalSndUpdateProgress(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid,
								 bool fc_skipped_xact);
static XLogRecPtr fc_WalSndWaitForWal(XLogRecPtr fc_loc);
static void fc_LagTrackerWrite(XLogRecPtr fc_lsn, TimestampTz fc_local_flush_time);
static TimeOffset fc_LagTrackerRead(int fc_head, XLogRecPtr fc_lsn, TimestampTz fc_now);
static bool fc_TransactionIdInRecentPast(TransactionId fc_xid, uint32 fc_epoch);

static void fc_WalSndSegmentOpen(XLogReaderState *fc_state, XLogSegNo fc_nextSegNo,
							  TimeLineID *fc_tli_p);


/* 在进入主命令循环之前初始化walsender进程 */
void InitWalSender(void)
{
	am_cascading_walsender = RecoveryInProgress();

	/* 在共享内存中为每个walsender创建一个数据结构 */
	fc_InitWalSenderSlot();

	/*
	 * 我们目前在walsender进程中不需要任何ResourceOwner，但是
	 * 如果需要，我们可以在这里调用CreateAuxProcessResourceOwner。
	 */

	/*
	 * 让postmaster知道我们是WAL发送器。一旦我们声明自己为
	 * WAL发送器进程，postmaster将允许我们在bgwriter之后生存
	 * 并在关闭序列中最后杀死我们，因此我们在关闭时获得了
	 * 流式传输所有剩余WAL的机会，包括关闭检查点。请注意，
	 * 没有回头路，我们在此之后不能写任何WAL记录。
	 */
	MarkPostmasterChildWalSender();
	SendPostmasterSignal(PMSIGNAL_ADVANCE_STATE_MACHINE);

	/*
	 * 如果客户端没有指定要连接的数据库，则在PGPROC中显示
	 * 我们宣传的xmin应影响所有数据库中的vacuum horizon。
	 * 这允许物理复制客户端发送热备用反馈，这将延迟
	 * 所有数据库中的vacuum清理。
	 */
	if (MyDatabaseId == InvalidOid)
	{
		Assert(MyProc->xmin == InvalidTransactionId);
		LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
		MyProc->statusFlags |= PROC_AFFECTS_ALL_HORIZONS;
		ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
		LWLockRelease(ProcArrayLock);
	}

	/* 为滞后跟踪初始化空时间戳缓冲区。 */
	lag_tracker = MemoryContextAllocZero(TopMemoryContext, sizeof(LagTracker));
}

/*
 * 出现错误后进行清理。
 *
 * WAL发送进程不像常规后端那样使用事务。
 * 此函数在WAL发送进程中发生错误后执行所需的清理，
 * 类似于常规后端中的事务中止所做的。
 */
void WalSndErrorCleanup(void)
{
	LWLockReleaseAll();
	ConditionVariableCancelSleep();
	pgstat_report_wait_end();

	if (xlogreader != NULL && xlogreader->seg.ws_file >= 0)
		wal_segment_close(xlogreader);

	if (MyReplicationSlot != NULL)
		ReplicationSlotRelease();

	ReplicationSlotCleanup();

	replication_active = false;

	/*
	 * 如果正在进行事务，它将清理我们的ResourceOwner，
	 * 但如果复制命令在没有事务的情况下设置了资源所有者，
	 * 我们现在必须清理它。
	 */
	if (!IsTransactionOrTransactionBlock())
		WalSndResourceCleanup(false);

	if (got_STOPPING || got_SIGUSR2)
		proc_exit(0);

	/* 恢复到启动状态 */
	WalSndSetState(WALSNDSTATE_STARTUP);
}

/*
 * 清理我们创建的任何 ResourceOwner。
 */
void WalSndResourceCleanup(bool fc_isCommit)
{
	ResourceOwner fc_resowner;

	if (CurrentResourceOwner == NULL)
		return;

	/*
	 * 不允许删除 CurrentResourceOwner，因此我们必须在本地变量中保存一个指针并首先清除它。
	 */
	fc_resowner = CurrentResourceOwner;
	CurrentResourceOwner = NULL;

	/* 现在我们可以释放资源并删除它。 */
	ResourceOwnerRelease(fc_resowner,
						 RESOURCE_RELEASE_BEFORE_LOCKS, fc_isCommit, true);
	ResourceOwnerRelease(fc_resowner,
						 RESOURCE_RELEASE_LOCKS, fc_isCommit, true);
	ResourceOwnerRelease(fc_resowner,
						 RESOURCE_RELEASE_AFTER_LOCKS, fc_isCommit, true);
	ResourceOwnerDelete(fc_resowner);
}

/*
 * 以有序的方式处理客户端连接中止。
 */
static void fc_WalSndShutdown(void)
{
	/*
	 * 重置 whereToSendOutput 以防止 ereport 尝试向备用发送更多消息。
	 */
	if (whereToSendOutput == DestRemote)
		whereToSendOutput = DestNone;

	proc_exit(0);
	abort();					/* 让编译器安静 */
}

/*
 * 处理 IDENTIFY_SYSTEM 命令。
 */
static void fc_IdentifySystem(void)
{
	char		fc_sysid[32];
	char		fc_xloc[MAXFNAMELEN];
	XLogRecPtr	fc_logptr;
	char	   *fc_dbname = NULL;
	DestReceiver *fc_dest;
	TupOutputState *fc_tstate;
	TupleDesc	fc_tupdesc;
	Datum		fc_values[4];
	bool		fc_nulls[4];
	TimeLineID	fc_currTLI;

	/*
	 * 回复一个包含一行四列的结果集。第一列是系统 ID，第二列是时间线 ID，第三列是当前 xlog 位置，第四列包含数据库名称（如果我们连接到某个数据库）。
	 */

	snprintf(fc_sysid, sizeof(fc_sysid), UINT64_FORMAT,
			 GetSystemIdentifier());

	am_cascading_walsender = RecoveryInProgress();
	if (am_cascading_walsender)
		fc_logptr = fc_GetStandbyFlushRecPtr(&fc_currTLI);
	else
		fc_logptr = GetFlushRecPtr(&fc_currTLI);

	snprintf(fc_xloc, sizeof(fc_xloc), "%X/%X", LSN_FORMAT_ARGS(fc_logptr));

	if (MyDatabaseId != InvalidOid)
	{
		MemoryContext fc_cur = CurrentMemoryContext;

		/* syscache 访问需要一个事务环境。 */
		StartTransactionCommand();
		/* 使 dbname 在 TX 上下文之外处于活动状态 */
		MemoryContextSwitchTo(fc_cur);
		fc_dbname = get_database_name(MyDatabaseId);
		CommitTransactionCommand();
		/* CommitTransactionCommand 切换到 TopMemoryContext */
		MemoryContextSwitchTo(fc_cur);
	}

	fc_dest = CreateDestReceiver(DestRemoteSimple);
	MemSet(fc_nulls, false, sizeof(fc_nulls));

	/* 需要一个表示四列的元组描述符 */
	fc_tupdesc = CreateTemplateTupleDesc(4);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 1, "systemid",
							  TEXTOID, -1, 0);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 2, "timeline",
							  INT4OID, -1, 0);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 3, "xlogpos",
							  TEXTOID, -1, 0);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 4, "dbname",
							  TEXTOID, -1, 0);

	/*
 * show_config_by_name_missing_ok - 等同于 SHOW X 命令，但实现为
 * 一个函数。如果 X 不存在，抑制错误并仅在 missing_ok 为真时返回 NULL。
 */
	fc_tstate = begin_tup_output_tupdesc(fc_dest, fc_tupdesc, &TTSOpsVirtual);

	/* 列 1：系统标识符 */
	fc_values[0] = CStringGetTextDatum(fc_sysid);

	/* 列 2：时间线 */
	fc_values[1] = Int32GetDatum(fc_currTLI);

	/* 列 3：wal 位置 */
	fc_values[2] = CStringGetTextDatum(fc_xloc);

	/* 列 4：数据库名称，如果没有则为 NULL */
	if (fc_dbname)
		fc_values[3] = CStringGetTextDatum(fc_dbname);
	else
		fc_nulls[3] = true;

	/* 为跨调用持久性创建一个函数上下文 */
	do_tup_output(fc_tstate, fc_values, fc_nulls);

	end_tup_output(fc_tstate);
}

/* 处理 READ_REPLICATION_SLOT 命令 */
static void fc_ReadReplicationSlot(ReadReplicationSlotCmd *fc_cmd)
{
#define READ_REPLICATION_SLOT_COLS 3
	ReplicationSlot *fc_slot;
	DestReceiver *fc_dest;
	TupOutputState *fc_tstate;
	TupleDesc	fc_tupdesc;
	Datum		fc_values[READ_REPLICATION_SLOT_COLS];
	bool		fc_nulls[READ_REPLICATION_SLOT_COLS];

	fc_tupdesc = CreateTemplateTupleDesc(READ_REPLICATION_SLOT_COLS);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 1, "slot_type",
							  TEXTOID, -1, 0);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 2, "restart_lsn",
							  TEXTOID, -1, 0);
	/* TimeLineID 是无符号的，因此 int4 不够宽。 */
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 3, "restart_tli",
							  INT8OID, -1, 0);

	MemSet(fc_values, 0, READ_REPLICATION_SLOT_COLS * sizeof(Datum));
	MemSet(fc_nulls, true, READ_REPLICATION_SLOT_COLS * sizeof(bool));

	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
	fc_slot = SearchNamedReplicationSlot(fc_cmd->slotname, false);
	if (fc_slot == NULL || !fc_slot->in_use)
	{
		LWLockRelease(ReplicationSlotControlLock);
	}
	else
	{
		ReplicationSlot fc_slot_contents;
		int			fc_i = 0;

		/* 在持有自旋锁的同时复制插槽内容 */
		SpinLockAcquire(&fc_slot->mutex);
		fc_slot_contents = *fc_slot;
		SpinLockRelease(&fc_slot->mutex);
		LWLockRelease(ReplicationSlotControlLock);

		if (OidIsValid(fc_slot_contents.data.database))
			ereport(ERROR,
					errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					errmsg("cannot use %s with a logical replication slot",
						   "READ_REPLICATION_SLOT"));

		/* 插槽类型 */
		fc_values[fc_i] = CStringGetTextDatum("physical");
		fc_nulls[fc_i] = false;
		fc_i++;

		/* 启动 LSN */
		if (!XLogRecPtrIsInvalid(fc_slot_contents.data.restart_lsn))
		{
			char		fc_xloc[64];

			snprintf(fc_xloc, sizeof(fc_xloc), "%X/%X",
					 LSN_FORMAT_ARGS(fc_slot_contents.data.restart_lsn));
			fc_values[fc_i] = CStringGetTextDatum(fc_xloc);
			fc_nulls[fc_i] = false;
		}
		fc_i++;

		/* 此 WAL 生产的时间线 */
		if (!XLogRecPtrIsInvalid(fc_slot_contents.data.restart_lsn))
		{
			TimeLineID	fc_slots_position_timeline;
			TimeLineID	fc_current_timeline;
			List	   *fc_timeline_history = NIL;

			/*
			 * 在恢复期间，使用当前正在重放的时间线来获取 LSN 位置的历史记录。
			 */
			if (RecoveryInProgress())
				(void) GetXLogReplayRecPtr(&fc_current_timeline);
			else
				fc_current_timeline = GetWALInsertionTimeLine();

			fc_timeline_history = readTimeLineHistory(fc_current_timeline);
			fc_slots_position_timeline = tliOfPointInHistory(fc_slot_contents.data.restart_lsn,
														  fc_timeline_history);
			fc_values[fc_i] = Int64GetDatum((int64) fc_slots_position_timeline);
			fc_nulls[fc_i] = false;
		}
		fc_i++;

		Assert(fc_i == READ_REPLICATION_SLOT_COLS);
	}

	fc_dest = CreateDestReceiver(DestRemoteSimple);
	fc_tstate = begin_tup_output_tupdesc(fc_dest, fc_tupdesc, &TTSOpsVirtual);
	do_tup_output(fc_tstate, fc_values, fc_nulls);
	end_tup_output(fc_tstate);
}


/*
 * 处理 TIMELINE_HISTORY 命令。
 */
static void fc_SendTimeLineHistory(TimeLineHistoryCmd *fc_cmd)
{
	StringInfoData fc_buf;
	char		fc_histfname[MAXFNAMELEN];
	char		fc_path[MAXPGPATH];
	int			fc_fd;
	off_t		fc_histfilelen;
	off_t		fc_bytesleft;
	Size		fc_len;

	/*
	 * 回复一个包含一行两列的结果集。第一列是历史文件的名称，第二列是内容。
	 */

	TLHistoryFileName(fc_histfname, fc_cmd->timeline);
	TLHistoryFilePath(fc_path, fc_cmd->timeline);

	/* 发送 RowDescription 消息 */
	pq_beginmessage(&fc_buf, 'T');
	pq_sendint16(&fc_buf, 2);		/* 2个字段 */

	/* 第一个字段 */
	pq_sendstring(&fc_buf, "filename");	/* col name */
	pq_sendint32(&fc_buf, 0);		/* 表 oid */
	pq_sendint16(&fc_buf, 0);		/* attnum */
	pq_sendint32(&fc_buf, TEXTOID);	/* 类型 oid */
	pq_sendint16(&fc_buf, -1);		/* typlen */
	pq_sendint32(&fc_buf, 0);		/* typmod */
	pq_sendint16(&fc_buf, 0);		/* 格式代码 */

	/* 第二个字段 */
	pq_sendstring(&fc_buf, "content"); /* col name */
	pq_sendint32(&fc_buf, 0);		/* 表 oid */
	pq_sendint16(&fc_buf, 0);		/* attnum */
	pq_sendint32(&fc_buf, TEXTOID);	/* 类型 oid */
	pq_sendint16(&fc_buf, -1);		/* typlen */
	pq_sendint32(&fc_buf, 0);		/* typmod */
	pq_sendint16(&fc_buf, 0);		/* 格式代码 */
	pq_endmessage(&fc_buf);

	/* 发送 DataRow 消息 */
	pq_beginmessage(&fc_buf, 'D');
	pq_sendint16(&fc_buf, 2);		/* 列数 */
	fc_len = strlen(fc_histfname);
	pq_sendint32(&fc_buf, fc_len);	/* col1 len */
	pq_sendbytes(&fc_buf, fc_histfname, fc_len);

	fc_fd = OpenTransientFile(fc_path, O_RDONLY | PG_BINARY);
	if (fc_fd < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not open file \"%s\": %m", fc_path)));

	/* 确定文件长度并将其发送给客户端 */
	fc_histfilelen = lseek(fc_fd, 0, SEEK_END);
	if (fc_histfilelen < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not seek to end of file \"%s\": %m", fc_path)));
	if (lseek(fc_fd, 0, SEEK_SET) != 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not seek to beginning of file \"%s\": %m", fc_path)));

	pq_sendint32(&fc_buf, fc_histfilelen);	/* col2 len */

	fc_bytesleft = fc_histfilelen;
	while (fc_bytesleft > 0)
	{
		PGAlignedBlock fc_rbuf;
		int			fc_nread;

		pgstat_report_wait_start(WAIT_EVENT_WALSENDER_TIMELINE_HISTORY_READ);
		fc_nread = read(fc_fd, fc_rbuf.data, sizeof(fc_rbuf));
		pgstat_report_wait_end();
		if (fc_nread < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read file \"%s\": %m",
							fc_path)));
		else if (fc_nread == 0)
			ereport(ERROR,
					(errcode(ERRCODE_DATA_CORRUPTED),
					 errmsg("could not read file \"%s\": read %d of %zu",
							fc_path, fc_nread, (Size) fc_bytesleft)));

		pq_sendbytes(&fc_buf, fc_rbuf.data, fc_nread);
		fc_bytesleft -= fc_nread;
	}

	if (CloseTransientFile(fc_fd) != 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not close file \"%s\": %m", fc_path)));

	pq_endmessage(&fc_buf);
}

/*
 * 处理 START_REPLICATION 命令。
 *
 * 目前，这个命令永远不会返回，但一个 ereport(ERROR) 会将我们带回主循环。
 */
static void fc_StartReplication(StartReplicationCmd *fc_cmd)
{
	StringInfoData fc_buf;
	XLogRecPtr	fc_FlushPtr;
	TimeLineID	fc_FlushTLI;

	/* 创建物理复制的xlogreader */
	xlogreader =
		XLogReaderAllocate(wal_segment_size, NULL,
						   XL_ROUTINE(.segment_open = fc_WalSndSegmentOpen,
									  .segment_close = wal_segment_close),
						   NULL);

	if (!xlogreader)
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("out of memory"),
				 errdetail("Failed while allocating a WAL reading processor.")));

	/*
	 * 在这里，我们假设我们在WAL中记录了足够的信息以进行
	 * 日志传输，因为这在PostmasterMain()中进行了检查。
	 *
	 * 注意：wal_level只能在关机时更改，因此在大多数情况下，
	 * 难以存在我们仍然可以看到的在wal_level='minimal'下写入的WAL数据。
	 */

	if (fc_cmd->slotname)
	{
		ReplicationSlotAcquire(fc_cmd->slotname, true);
		if (SlotIsLogical(MyReplicationSlot))
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("cannot use a logical replication slot for physical replication")));

		/*
		 * 我们不需要在这里验证槽的restart_lsn；相反，我们依赖调用者请求使用的起始点。
		 * 如果WAL段不存在，我们稍后会失败。
		 */
	}

	/*
	 * 选择时间线。如果客户端明确给出了它，请使用
	 * 它。否则，使用最后一个重放记录的时间线。
	 */
	am_cascading_walsender = RecoveryInProgress();
	if (am_cascading_walsender)
		fc_FlushPtr = fc_GetStandbyFlushRecPtr(&fc_FlushTLI);
	else
		fc_FlushPtr = GetFlushRecPtr(&fc_FlushTLI);

	if (fc_cmd->timeline != 0)
	{
		XLogRecPtr	fc_switchpoint;

		sendTimeLine = fc_cmd->timeline;
		if (sendTimeLine == fc_FlushTLI)
		{
			sendTimeLineIsHistoric = false;
			sendTimeLineValidUpto = InvalidXLogRecPtr;
		}
		else
		{
			List	   *fc_timeLineHistory;

			sendTimeLineIsHistoric = true;

			/*
			 * 检查客户端请求的时间线是否存在，且
			 * 请求的起始位置是否在该时间线上。
			 */
			fc_timeLineHistory = readTimeLineHistory(fc_FlushTLI);
			fc_switchpoint = tliSwitchPoint(fc_cmd->timeline, fc_timeLineHistory,
										 &sendTimeLineNextTLI);
			list_free_deep(fc_timeLineHistory);

			/*
			 * 在历史记录中找到了请求的时间线。检查
			 * 请求的起始点是否在我们历史记录中的该时间线上。
			 *
			 * 这是故意很宽松的。我们只检查在切换点之前
			 * 我们没有从请求的时间线分支出来。我们不检查
			 * 在请求的起始点之前是否切换*到*它。这是因为客户端可以合法地
			 * 请求从包含切换点的WAL段的开头开始复制，但在新的时间线中，这样
			 * 不会得到一个部分段。如果您请求一个太旧的起始点，
			 * 当我们无法在pg_wal中找到请求的WAL段时，会出现错误。
			 *
			 * XXX：我们可以在这里更严格，只允许一个比切换点
			 * 旧的起始点，如果它仍在同一个WAL段中。
			 */
			if (!XLogRecPtrIsInvalid(fc_switchpoint) &&
				fc_switchpoint < fc_cmd->startpoint)
			{
				ereport(ERROR,
						(errmsg("requested starting point %X/%X on timeline %u is not in this server's history",
								LSN_FORMAT_ARGS(fc_cmd->startpoint),
								fc_cmd->timeline),
						 errdetail("This server's history forked from timeline %u at %X/%X.",
								   fc_cmd->timeline,
								   LSN_FORMAT_ARGS(fc_switchpoint))));
			}
			sendTimeLineValidUpto = fc_switchpoint;
		}
	}
	else
	{
		sendTimeLine = fc_FlushTLI;
		sendTimeLineValidUpto = InvalidXLogRecPtr;
		sendTimeLineIsHistoric = false;
	}

	streamingDoneSending = streamingDoneReceiving = false;

	/* 如果没有任何内容可流，则甚至不进入COPY模式 */
	if (!sendTimeLineIsHistoric || fc_cmd->startpoint < sendTimeLineValidUpto)
	{
		/*
		 * 当我们首次启动复制时，备用服务器将落后于
		 * 主服务器。对于某些应用程序，例如同步
		 * 复制，明确的初始
		 * 追赶模式状态很重要，这样我们就可以在后续更改流状态时触发操作。
		 * 我们可能会在这个状态下停留很长时间，这正是我们希望能够监控我们是否
		 * 仍然留在这里的原因。
		 */
		WalSndSetState(WALSNDSTATE_CATCHUP);

		/* 发送CopyBothResponse消息，并开始流 */
		pq_beginmessage(&fc_buf, 'W');
		pq_sendbyte(&fc_buf, 0);
		pq_sendint16(&fc_buf, 0);
		pq_endmessage(&fc_buf);
		pq_flush();

		/*
		 * 不允许从尚未在此服务器上刷新到
		 * 磁盘的WAL的未来点流。
		 */
		if (fc_FlushPtr < fc_cmd->startpoint)
		{
			ereport(ERROR,
					(errmsg("requested starting point %X/%X is ahead of the WAL flush position of this server %X/%X",
							LSN_FORMAT_ARGS(fc_cmd->startpoint),
							LSN_FORMAT_ARGS(fc_FlushPtr))));
		}

		/* 从请求的点开始流 */
		sentPtr = fc_cmd->startpoint;

		/* 共享内存状态也初始化 */
		SpinLockAcquire(&MyWalSnd->mutex);
		MyWalSnd->sentPtr = sentPtr;
		SpinLockRelease(&MyWalSnd->mutex);
#ifdef FDD
		SyncRepInitConfig(false);

#else
		SyncRepInitConfig();
#endif

		/* walsender的主循环 */
		replication_active = true;

		fc_WalSndLoop(fc_XLogSendPhysical);

		replication_active = false;
		if (got_STOPPING)
			proc_exit(0);
		WalSndSetState(WALSNDSTATE_STARTUP);

		Assert(streamingDoneSending && streamingDoneReceiving);
	}

	if (fc_cmd->slotname)
		ReplicationSlotRelease();

	/*
	 * 复制现在完成。发送一个单行结果集，指示下一个
	 * 时间线。
	 */
	if (sendTimeLineIsHistoric)
	{
		char		fc_startpos_str[8 + 1 + 8 + 1];
		DestReceiver *fc_dest;
		TupOutputState *fc_tstate;
		TupleDesc	fc_tupdesc;
		Datum		fc_values[2];
		bool		fc_nulls[2];

		snprintf(fc_startpos_str, sizeof(fc_startpos_str), "%X/%X",
				 LSN_FORMAT_ARGS(sendTimeLineValidUpto));

		fc_dest = CreateDestReceiver(DestRemoteSimple);
		MemSet(fc_nulls, false, sizeof(fc_nulls));

		/*
		 * 需要一个表示两列的元组描述符。int8可能是
		 * 对于此而言一个令人惊讶的数据类型，但理论上int4对此并不宽，因为TimeLineID是无符号的。
		 */
		fc_tupdesc = CreateTemplateTupleDesc(2);
		TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 1, "next_tli",
								  INT8OID, -1, 0);
		TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 2, "next_tli_startpos",
								  TEXTOID, -1, 0);

		/* 为元组投影做准备 */
		fc_tstate = begin_tup_output_tupdesc(fc_dest, fc_tupdesc, &TTSOpsVirtual);

		fc_values[0] = Int64GetDatum((int64) sendTimeLineNextTLI);
		fc_values[1] = CStringGetTextDatum(fc_startpos_str);

		/* 为跨调用持久性创建一个函数上下文 */
		do_tup_output(fc_tstate, fc_values, fc_nulls);

		end_tup_output(fc_tstate);
	}

	/* 发送CommandComplete消息 */
	EndReplicationCommand("START_STREAMING");
}

/*
 * XLogReaderRoutine->page_read回调，用于逻辑解码上下文，作为一个
 * walsender进程。
 *
 * 在walsender内部，我们可以做得比read_local_xlog_page更好，
 * 它必须进行普通的休眠/繁忙循环，因为walsender的闩锁在每次WAL被刷新时
 * 被设置。
 */
static int fc_logical_read_xlog_page(XLogReaderState *fc_state, XLogRecPtr fc_targetPagePtr, int fc_reqLen,
					   XLogRecPtr fc_targetRecPtr, char *fc_cur_page)
{
	XLogRecPtr	fc_flushptr;
	int			fc_count;
	WALReadError fc_errinfo;
	XLogSegNo	fc_segno;
	TimeLineID	fc_currTLI = GetWALInsertionTimeLine();

	/*
	 * 由于逻辑解码仅在主服务器上允许，我们知道
	 * 当前的时间线ID不能再变化。如果我们在备用服务器上执行此操作，
	 * 我们将必须担心这里计算的值因提升或时间线更改而失效。
	 */
	XLogReadDetermineTimeline(fc_state, fc_targetPagePtr, fc_reqLen, fc_currTLI);
	sendTimeLineIsHistoric = (fc_state->currTLI != fc_currTLI);
	sendTimeLine = fc_state->currTLI;
	sendTimeLineValidUpto = fc_state->currTLIValidUntil;
	sendTimeLineNextTLI = fc_state->nextTLI;

	/* 确保我们有足够的WAL可用 */
	fc_flushptr = fc_WalSndWaitForWal(fc_targetPagePtr + fc_reqLen);

	/* 如果没有则失败（意味着我们将关闭） */
	if (fc_flushptr < fc_targetPagePtr + fc_reqLen)
		return -1;

	if (fc_targetPagePtr + XLOG_BLCKSZ <= fc_flushptr)
		fc_count = XLOG_BLCKSZ;	/* 可用多个块 */
	else
		fc_count = fc_flushptr - fc_targetPagePtr;	/* 页面的一部分可用 */

	/* 现在实际读取数据，我们知道它在那里 */
	if (!WALRead(fc_state,
				 fc_cur_page,
				 fc_targetPagePtr,
				 XLOG_BLCKSZ,
				 fc_state->seg.ws_tli, /* 传递当前TLI，因为只有
									 * WalSndSegmentOpen 控制是否需要新的
									 * TLI。 */
				 &fc_errinfo))
		WALReadRaiseError(&fc_errinfo);

	/*
	 * 在读取到缓冲区后，检查我们读取的内容是否有效。我们在
	 * 读取后执行此操作，因为即使在我们打开它时段存在，
	 * 在我们读取期间它可能会被回收或删除。read()在这种情况下成功，
	 * 但我们尝试读取的数据可能已经被新WAL记录覆盖。
	 */
	XLByteToSeg(fc_targetPagePtr, fc_segno, fc_state->segcxt.ws_segsize);
	CheckXLogRemoved(fc_segno, fc_state->seg.ws_tli);

	return fc_count;
}

/*
 * 处理提供给CREATE_REPLICATION_SLOT的额外选项。
 */
static void fc_parseCreateReplSlotOptions(CreateReplicationSlotCmd *fc_cmd,
						   bool *fc_reserve_wal,
						   CRSSnapshotAction *fc_snapshot_action,
						   bool *fc_two_phase)
{
	ListCell   *fc_lc;
	bool		fc_snapshot_action_given = false;
	bool		fc_reserve_wal_given = false;
	bool		fc_two_phase_given = false;

	/* 解析选项 */
	foreach(fc_lc, fc_cmd->options)
	{
		DefElem    *fc_defel = (DefElem *) lfirst(fc_lc);

		if (strcmp(fc_defel->defname, "snapshot") == 0)
		{
			char	   *fc_action;

			if (fc_snapshot_action_given || fc_cmd->kind != REPLICATION_KIND_LOGICAL)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));

			fc_action = defGetString(fc_defel);
			fc_snapshot_action_given = true;

			if (strcmp(fc_action, "export") == 0)
				*fc_snapshot_action = CRS_EXPORT_SNAPSHOT;
			else if (strcmp(fc_action, "nothing") == 0)
				*fc_snapshot_action = CRS_NOEXPORT_SNAPSHOT;
			else if (strcmp(fc_action, "use") == 0)
				*fc_snapshot_action = CRS_USE_SNAPSHOT;
			else
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("unrecognized value for CREATE_REPLICATION_SLOT option \"%s\": \"%s\"",
								fc_defel->defname, fc_action)));
		}
		else if (strcmp(fc_defel->defname, "reserve_wal") == 0)
		{
			if (fc_reserve_wal_given || fc_cmd->kind != REPLICATION_KIND_PHYSICAL)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));

			fc_reserve_wal_given = true;
			*fc_reserve_wal = defGetBoolean(fc_defel);
		}
		else if (strcmp(fc_defel->defname, "two_phase") == 0)
		{
			if (fc_two_phase_given || fc_cmd->kind != REPLICATION_KIND_LOGICAL)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));
			fc_two_phase_given = true;
			*fc_two_phase = defGetBoolean(fc_defel);
		}
		else
			elog(ERROR, "unrecognized option: %s", fc_defel->defname);
	}
}

/*
 * 创建一个新的复制槽。
 */
static void fc_CreateReplicationSlot(CreateReplicationSlotCmd *fc_cmd)
{
	const char *fc_snapshot_name = NULL;
	char		fc_xloc[MAXFNAMELEN];
	char	   *fc_slot_name;
	bool		fc_reserve_wal = false;
	bool		fc_two_phase = false;
	CRSSnapshotAction fc_snapshot_action = CRS_EXPORT_SNAPSHOT;
	DestReceiver *fc_dest;
	TupOutputState *fc_tstate;
	TupleDesc	fc_tupdesc;
	Datum		fc_values[4];
	bool		fc_nulls[4];

	Assert(!MyReplicationSlot);

	fc_parseCreateReplSlotOptions(fc_cmd, &fc_reserve_wal, &fc_snapshot_action, &fc_two_phase);

	if (fc_cmd->kind == REPLICATION_KIND_PHYSICAL)
	{
		ReplicationSlotCreate(fc_cmd->slotname, false,
							  fc_cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
#ifdef FDD
							  false, fc_cmd->failover);
#else
							  false);
#endif
	}
	else
	{
		CheckLogicalDecodingRequirements();

		/*
		 * 初始创建持久槽为瞬态 - 这使我们能够在初始化期间
		 * 良好地处理错误，因为如果该事务失败，它将被丢弃。
		 * 最后我们将使其持久。临时槽可以从一开始就创建为
		 * 临时，因为它们在出错时也会被丢弃。
		 */
		ReplicationSlotCreate(fc_cmd->slotname, true,
							  fc_cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
#ifdef FDD
							  fc_two_phase, fc_cmd->failover);
#else
							  fc_two_phase);
#endif

	}

	if (fc_cmd->kind == REPLICATION_KIND_LOGICAL)
	{
		LogicalDecodingContext *fc_ctx;
		bool		fc_need_full_snapshot = false;

		/*
		 * 及早进行选项检查，以便在调用可能需要较长时间的
		 * DecodingContextFindStartpoint 之前退出。
		 */
		if (fc_snapshot_action == CRS_EXPORT_SNAPSHOT)
		{
			if (IsTransactionBlock())
				ereport(ERROR,
				/*- translator: %s 是一个 CREATE_REPLICATION_SLOT 语句 */
						(errmsg("%s must not be called inside a transaction",
								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')")));

			fc_need_full_snapshot = true;
		}
		else if (fc_snapshot_action == CRS_USE_SNAPSHOT)
		{
			if (!IsTransactionBlock())
				ereport(ERROR,
				/*- translator: %s 是一个 CREATE_REPLICATION_SLOT 语句 */
						(errmsg("%s must be called inside a transaction",
								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));

			if (XactIsoLevel != XACT_REPEATABLE_READ)
				ereport(ERROR,
				/*- translator: %s 是一个 CREATE_REPLICATION_SLOT 语句 */
						(errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));

			if (FirstSnapshotSet)
				ereport(ERROR,
				/*- translator: %s 是一个 CREATE_REPLICATION_SLOT 语句 */
						(errmsg("%s must be called before any query",
								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));

			if (IsSubTransaction())
				ereport(ERROR,
				/*- translator: %s 是一个 CREATE_REPLICATION_SLOT 语句 */
						(errmsg("%s must not be called in a subtransaction",
								"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));

			fc_need_full_snapshot = true;
		}

		fc_ctx = CreateInitDecodingContext(fc_cmd->plugin, NIL, fc_need_full_snapshot,
										InvalidXLogRecPtr,
										XL_ROUTINE(.page_read = fc_logical_read_xlog_page,
												   .segment_open = fc_WalSndSegmentOpen,
												   .segment_close = wal_segment_close),
										fc_WalSndPrepareWrite, fc_WalSndWriteData,
										fc_WalSndUpdateProgress);

		/*
		 * 表示我们不需要超时机制。我们只是
		 * 创建复制槽，尚未接受反馈
		 * 消息或发送心跳。由于我们可能需要等待
		 * 进一步的 WAL，walsender 可能会被过早杀死。
		 */
		last_reply_timestamp = 0;

		/* 构建初始快照，可能需要一些时间 */
		DecodingContextFindStartpoint(fc_ctx);

		/*
		 * 如果有人要求我们导出或使用快照。
		 *
		 * 注意。我们将在这样做时将 snapbuild.c 类型的快照转换为正常
		 * 快照。
		 */
		if (fc_snapshot_action == CRS_EXPORT_SNAPSHOT)
		{
			fc_snapshot_name = SnapBuildExportSnapshot(fc_ctx->snapshot_builder);
		}
		else if (fc_snapshot_action == CRS_USE_SNAPSHOT)
		{
			Snapshot	fc_snap;

			fc_snap = SnapBuildInitialSnapshot(fc_ctx->snapshot_builder);
			RestoreTransactionSnapshot(fc_snap, MyProc);
		}

		/* 不再需要解码上下文 */
		FreeDecodingContext(fc_ctx);

		if (!fc_cmd->temporary)
			ReplicationSlotPersist();
	}
	else if (fc_cmd->kind == REPLICATION_KIND_PHYSICAL && fc_reserve_wal)
	{
		ReplicationSlotReserveWal();

		ReplicationSlotMarkDirty();

		/* 如果这是一个永久槽，则将此槽写入磁盘。 */
		if (!fc_cmd->temporary)
			ReplicationSlotSave();
	}

	snprintf(fc_xloc, sizeof(fc_xloc), "%X/%X",
			 LSN_FORMAT_ARGS(MyReplicationSlot->data.confirmed_flush));

	fc_dest = CreateDestReceiver(DestRemoteSimple);
	MemSet(fc_nulls, false, sizeof(fc_nulls));

	/*----------
	 * 需要一个表示四列的元组描述符：
	 * - 第一列：槽名称
	 * - 第二列：我们变得一致时的 LSN
	 * - 第三列：导出快照的名称
	 * - 第四列：输出插件
	 *----------
	 */
	fc_tupdesc = CreateTemplateTupleDesc(4);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 1, "slot_name",
							  TEXTOID, -1, 0);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 2, "consistent_point",
							  TEXTOID, -1, 0);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 3, "snapshot_name",
							  TEXTOID, -1, 0);
	TupleDescInitBuiltinEntry(fc_tupdesc, (AttrNumber) 4, "output_plugin",
							  TEXTOID, -1, 0);

	/*
 * show_config_by_name_missing_ok - 等同于 SHOW X 命令，但实现为
 * 一个函数。如果 X 不存在，抑制错误并仅在 missing_ok 为真时返回 NULL。
 */
	fc_tstate = begin_tup_output_tupdesc(fc_dest, fc_tupdesc, &TTSOpsVirtual);

	/* slot_name */
	fc_slot_name = NameStr(MyReplicationSlot->data.name);
	fc_values[0] = CStringGetTextDatum(fc_slot_name);

	/* 一致的 WAL 位置 */
	fc_values[1] = CStringGetTextDatum(fc_xloc);

	/* 快照名称，如果没有则为 NULL */
	if (fc_snapshot_name != NULL)
		fc_values[2] = CStringGetTextDatum(fc_snapshot_name);
	else
		fc_nulls[2] = true;

	/* 插件，如果没有则为 NULL */
	if (fc_cmd->plugin != NULL)
		fc_values[3] = CStringGetTextDatum(fc_cmd->plugin);
	else
		fc_nulls[3] = true;

	/* 为跨调用持久性创建一个函数上下文 */
	do_tup_output(fc_tstate, fc_values, fc_nulls);
	end_tup_output(fc_tstate);

	ReplicationSlotRelease();
}

/*
 * 处理一个不再需要的复制槽。
 */
static void fc_DropReplicationSlot(DropReplicationSlotCmd *fc_cmd)
{
	ReplicationSlotDrop(fc_cmd->slotname, !fc_cmd->wait);
}

/*
 * 加载先前初始化的逻辑槽并准备发送数据（通过
 * WalSndLoop）。
 */
static void fc_StartLogicalReplication(StartReplicationCmd *fc_cmd)
{
	StringInfoData fc_buf;
	QueryCompletion fc_qc;

	/* 确保我们的要求仍然满足 */
	CheckLogicalDecodingRequirements();

	Assert(!MyReplicationSlot);

	ReplicationSlotAcquire(fc_cmd->slotname, true);

	if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot read from logical replication slot \"%s\"",
						fc_cmd->slotname),
				 errdetail("This slot has been invalidated because it exceeded the maximum reserved size.")));

	/*
	 * 强制断开连接，以便解码代码不需要关心
	 * 从恢复运行到正常环境运行的最终切换。客户端代码需要处理重连。
	 */
	if (am_cascading_walsender && !RecoveryInProgress())
	{
		ereport(LOG,
				(errmsg("terminating walsender process after promotion")));
		got_STOPPING = true;
	}

	/*
	 * 创建我们的解码上下文，使其从先前确认的位置开始。
	 *
	 * 在发送 CopyBothResponse 消息之前执行此操作，以便及时报告任何错误。
	 */
	logical_decoding_ctx =
		CreateDecodingContext(fc_cmd->startpoint, fc_cmd->options, false,
							  XL_ROUTINE(.page_read = fc_logical_read_xlog_page,
										 .segment_open = fc_WalSndSegmentOpen,
										 .segment_close = wal_segment_close),
							  fc_WalSndPrepareWrite, fc_WalSndWriteData,
							  fc_WalSndUpdateProgress);
	xlogreader = logical_decoding_ctx->reader;
#ifdef FDD
	//故障倒换逻辑复制槽延迟commit消息发送，需要的回调函数
	logical_decoding_ctx->fdb_keepalive_cb = fc_FDB_Wal_SndKeepalive;
	logical_decoding_ctx->failover_logical = MyReplicationSlot->data.failover;
#endif
	WalSndSetState(WALSNDSTATE_CATCHUP);

	/* 发送CopyBothResponse消息，并开始流 */
	pq_beginmessage(&fc_buf, 'W');
	pq_sendbyte(&fc_buf, 0);
	pq_sendint16(&fc_buf, 0);
	pq_endmessage(&fc_buf);
	pq_flush();

	/* 从所需的最旧 WAL 开始读取。 */
	XLogBeginRead(logical_decoding_ctx->reader,
				  MyReplicationSlot->data.restart_lsn);

	/*
	 * 报告我们将发送进一步提交的位置信息为当前 sentPtr。
	 */
	sentPtr = MyReplicationSlot->data.confirmed_flush;

	/* 还更新共享内存中的发送位置状态 */
	SpinLockAcquire(&MyWalSnd->mutex);
	MyWalSnd->sentPtr = MyReplicationSlot->data.restart_lsn;
	SpinLockRelease(&MyWalSnd->mutex);

	replication_active = true;

#ifdef FDD
	SyncRepInitConfig(MyReplicationSlot->data.failover);
#else
	SyncRepInitConfig();
#endif

	/* walsender的主循环 */
	fc_WalSndLoop(fc_XLogSendLogical);

	FreeDecodingContext(logical_decoding_ctx);
	ReplicationSlotRelease();

	replication_active = false;
	if (got_STOPPING)
		proc_exit(0);
	WalSndSetState(WALSNDSTATE_STARTUP);

	/* 退出 COPY 模式（CommandComplete）。 */
	SetQueryCompletion(&fc_qc, CMDTAG_COPY, 0);
	EndCommand(&fc_qc, DestRemote, false);
}

/*
 * LogicalDecodingContext 'prepare_write' 回调。
 *
 * 准备写入 StringInfo。
 *
 * 不要在这里做任何持久性的事情，因为数据可能不会被处理。
 */
static void fc_WalSndPrepareWrite(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid, bool fc_last_write)
{
	/* 不能通过多次发送相同的 LSN 让同步复制感到困惑 */
	if (!fc_last_write)
		fc_lsn = InvalidXLogRecPtr;

	resetStringInfo(fc_ctx->out);

	pq_sendbyte(fc_ctx->out, 'w');
	pq_sendint64(fc_ctx->out, fc_lsn);	/* dataStart */
	pq_sendint64(fc_ctx->out, fc_lsn);	/* walEnd */

	/*
	 * 稍后填充发送时间，就像在 XLogSendPhysical 中所做的那样，但
	 * 在这里保留空间。
	 */
	pq_sendint64(fc_ctx->out, 0);	/* sendtime */
}

/*
 * LogicalDecodingContext 'write' 回调。
 *
 * 实际将之前由 WalSndPrepareWrite 准备好的数据写出到
 * 网络。根据需要花费时间，但在此期间处理来自另一方的回复并检查超时。
 */
static void fc_WalSndWriteData(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid,
				bool fc_last_write)
{
	TimestampTz fc_now;

	/*
	 * 最后填充发送时间戳，以便尽可能晚地处理。
	 * 这有点丑陋，但协议已经设置，因为它已由
	 * 多个版本通过流式物理复制使用。
	 */
	resetStringInfo(&tmpbuf);
	fc_now = GetCurrentTimestamp();
	pq_sendint64(&tmpbuf, fc_now);
	memcpy(&fc_ctx->out->data[1 + sizeof(int64) + sizeof(int64)],
		   tmpbuf.data, sizeof(int64));

	/* 在 CopyData 数据包中输出先前收集的数据 */
	pq_putmessage_noblock('d', fc_ctx->out->data, fc_ctx->out->len);

	CHECK_FOR_INTERRUPTS();

	/* 尝试将待处理输出刷新到客户端 */
	if (pq_flush_if_writable() != 0)
		fc_WalSndShutdown();

	/* 尝试采取快速路径，除非我们接近 walsender 超时。 */
	if (fc_now < TimestampTzPlusMilliseconds(last_reply_timestamp,
										  wal_sender_timeout / 2) &&
		!pq_is_send_pending())
	{
		return;
	}

	/* 如果这里有待处理写入，则进入慢路径 */
	fc_ProcessPendingWrites();
}

/*
 * 等待直到没有待处理的写入。同时处理来自另一方的回复并检查超时。
 */
static void fc_ProcessPendingWrites(void)
{
	for (;;)
	{
		long		fc_sleeptime;

		/* 检查来自客户端的输入 */
		fc_ProcessRepliesIfAny();

		/* 超过超时时间则退出 */
		fc_WalSndCheckTimeOut();

		/* 如果时间到了则发送保持连接消息 */
		fc_WalSndKeepaliveIfNecessary();

		if (!pq_is_send_pending())
			break;

		fc_sleeptime = fc_WalSndComputeSleeptime(GetCurrentTimestamp());

		/* 睡眠直到发生某事或我们超时 */
		fc_WalSndWait(WL_SOCKET_WRITEABLE | WL_SOCKET_READABLE, fc_sleeptime,
				   WAIT_EVENT_WAL_SENDER_WRITE_DATA);

		/* 清除任何已经等待的唤醒 */
		ResetLatch(MyLatch);

		CHECK_FOR_INTERRUPTS();

		/* 处理最近收到的任何请求或信号 */
		if (ConfigReloadPending)
		{
			ConfigReloadPending = false;
			ProcessConfigFile(PGC_SIGHUP);
#ifdef FDD
			SyncRepInitConfig(MyReplicationSlot != NULL ? MyReplicationSlot->data.failover : false);
#else
			SyncRepInitConfig();
#endif
		}

		/* 尝试将待处理输出刷新到客户端 */
		if (pq_flush_if_writable() != 0)
			fc_WalSndShutdown();
	}

	/* 重新激活闩锁，以便WalSndLoop知道继续 */
	SetLatch(MyLatch);
}

/*
 * LogicalDecodingContext 'update_progress' 回调.
 *
 * 将当前位置写入延迟跟踪器（见XLogSendPhysical）。
 *
 * 在跳过空事务时，如果需要，则发送保持连接消息。
 */
static void fc_WalSndUpdateProgress(LogicalDecodingContext *fc_ctx, XLogRecPtr fc_lsn, TransactionId fc_xid,
					 bool fc_skipped_xact)
{
	static TimestampTz fc_sendTime = 0;
	TimestampTz fc_now = GetCurrentTimestamp();
	bool		fc_pending_writes = false;
	bool		fc_end_xact = fc_ctx->end_xact;

	/*
	 * 每个 WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 最多跟踪一次延迟，以
	 * 避免在我们频繁提交时淹没延迟跟踪器。
	 *
	 * 我们没有获取下游任何 LSN 确认的机制，除了事务结束 LSN。
	 * 所以，我们仅跟踪事务结束的 LSN。
	 */
#define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS	1000
	if (fc_end_xact && TimestampDifferenceExceeds(fc_sendTime, fc_now,
											   WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
	{
		fc_LagTrackerWrite(fc_lsn, fc_now);
		fc_sendTime = fc_now;
	}

	/*
	 * 在同步复制中跳过空事务时，我们发送一个保持连接消息以避免
	 * 延迟这些事务。
	 *
	 * 在这里检查sync_standbys_defined标志不需要加锁，因为在最坏的情况下，我们
	 * 只是发送一条额外的保持连接消息，而这在实际上并不需要。
	 */
	if (fc_skipped_xact &&
		SyncRepRequested() &&
		((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
	{
		fc_WalSndKeepalive(false, fc_lsn);

		/* 尝试将待处理输出刷新到客户端 */
		if (pq_flush_if_writable() != 0)
			fc_WalSndShutdown();

		/* 如果我们在这里有待处理的写入，确保它实际上已经刷新 */
		if (pq_is_send_pending())
			fc_pending_writes = true;
	}

	/*
	 * 如果有任何待处理的写入，则处理它们，或者尝试在需要时发送保持连接消息。
	 * 我们无需在事务结束时尝试发送保持连接消息，因为那将在稍后的时间点完成。
	 * 这仅对大型事务所需，此时我们不向下游发送任何更改，接收方可能会因此超时。
	 */
	if (fc_pending_writes || (!fc_end_xact &&
						   fc_now >= TimestampTzPlusMilliseconds(last_reply_timestamp,
															  wal_sender_timeout / 2)))
		fc_ProcessPendingWrites();
}

/*
 * 等待WAL < loc被刷新到磁盘，以便可以安全地发送给客户端。
 *
 * 返回已刷新的WAL的结束LSN。通常这将是 >= loc，但
 * 如果我们检测到关闭请求（无论是来自postmaster还是客户端），
 * 我们将提前返回，所以调用者必须始终检查。
 */
static XLogRecPtr fc_WalSndWaitForWal(XLogRecPtr fc_loc)
{
	int			fc_wakeEvents;
	static XLogRecPtr fc_RecentFlushPtr = InvalidXLogRecPtr;

	/*
	 * 快速路径以避免获取自旋锁，如果我们已经知道
	 * 有足够的WAL可用。如果我们落后得太多，这一点特别有趣。
	 */
	if (fc_RecentFlushPtr != InvalidXLogRecPtr &&
		fc_loc <= fc_RecentFlushPtr)
		return fc_RecentFlushPtr;

	/* 获取更最近的刷新指针。 */
	if (!RecoveryInProgress())
		fc_RecentFlushPtr = GetFlushRecPtr(NULL);
	else
		fc_RecentFlushPtr = GetXLogReplayRecPtr(NULL);

	for (;;)
	{
		long		fc_sleeptime;

		/* 清除任何已经等待的唤醒 */
		ResetLatch(MyLatch);

		CHECK_FOR_INTERRUPTS();

		/* 处理最近收到的任何请求或信号 */
		if (ConfigReloadPending)
		{
			ConfigReloadPending = false;
			ProcessConfigFile(PGC_SIGHUP);
#ifdef FDD
			SyncRepInitConfig(MyReplicationSlot != NULL ? MyReplicationSlot->data.failover : false);
#else
			SyncRepInitConfig();
#endif
		}

		/* 检查来自客户端的输入 */
		fc_ProcessRepliesIfAny();

		/*
		 * 如果我们正在关闭，请触发待处理的WAL被写出，
		 * 否则我们可能会一直等待从未写出的WAL，因为
		 * walwriter已经关闭。
		 */
		if (got_STOPPING)
			XLogBackgroundFlush();

		/* 更新我们对当前已刷出位置的认识。 */
		if (!RecoveryInProgress())
			fc_RecentFlushPtr = GetFlushRecPtr(NULL);
		else
			fc_RecentFlushPtr = GetXLogReplayRecPtr(NULL);

		/*
		 * 如果postmaster要求我们停止，就不要再等待了。
		 *
		 * 在重新计算RecentFlushPtr之后进行此检查很重要，
		 * 以便我们可以在关闭之前发送所有剩余数据。
		 */
		if (got_STOPPING)
			break;

		/*
		 * 我们仅向客户端发送完整解码事务的常规消息，
		 * 但同步复制和walsender关闭可能在等待较晚的位置。
		 * 所以，在睡眠之前，我们发送一个包含刷新位置的ping。
		 * 如果接收方闲置，则此保持连接将触发回复。
		 * 处理回复将更新这些MyWalSnd位置。
		 */
		if (MyWalSnd->flush < sentPtr &&
			MyWalSnd->write < sentPtr &&
			!waiting_for_ping_response)
			fc_WalSndKeepalive(false, InvalidXLogRecPtr);

		/* 检查我们是否完成 */
		if (fc_loc <= fc_RecentFlushPtr)
			break;

		/* 等待新的WAL。由于我们需要等待，我们现在已经跟上了。 */
		WalSndCaughtUp = true;

		/*
		 * 尝试将任何待处理的输出刷新到客户端。
		 */
		if (pq_flush_if_writable() != 0)
			fc_WalSndShutdown();

		/*
		 * 如果我们已经收到客户端的CopyDone，自己发送了CopyDone，
		 * 并且输出缓冲区为空，则是时候退出流式传输，
		 * 因此失败当前的WAL获取请求。
		 */
		if (streamingDoneReceiving && streamingDoneSending &&
			!pq_is_send_pending())
			break;

		/* 超过超时时间则退出 */
		fc_WalSndCheckTimeOut();

		/* 如果时间到了则发送保持连接消息 */
		fc_WalSndKeepaliveIfNecessary();

		/*
		 * 睡眠直到发生某事或我们超时。如果仍有待处理输出，
		 * 还要等待套接字变为可写。
		 * 否则，我们可能在等待新的WAL生成时，坐在可发送的输出数据上。
		 * （但如果我们没有要发送的内容，我们不想在套接字可写时唤醒。）
		 */
		fc_sleeptime = fc_WalSndComputeSleeptime(GetCurrentTimestamp());

		fc_wakeEvents = WL_SOCKET_READABLE;

		if (pq_is_send_pending())
			fc_wakeEvents |= WL_SOCKET_WRITEABLE;

		fc_WalSndWait(fc_wakeEvents, fc_sleeptime, WAIT_EVENT_WAL_SENDER_WAIT_WAL);
	}

	/* 重新激活闩锁，以便WalSndLoop知道继续 */
	SetLatch(MyLatch);
	return fc_RecentFlushPtr;
}

/*
 * 执行传入的复制命令。
 *
 * 如果cmd_string被识别为WalSender命令，则返回true，
 * 否则返回false。
 */
bool exec_replication_command(const char *fc_cmd_string)
{
	int			fc_parse_rc;
	Node	   *fc_cmd_node;
	const char *fc_cmdtag;
	MemoryContext fc_cmd_context;
	MemoryContext fc_old_context;

	/*
	 * 如果WAL发送方被告知关闭即将来临，则相应地切换其
	 * 状态，以正确处理下一个复制命令。
	 */
	if (got_STOPPING)
		WalSndSetState(WALSNDSTATE_STOPPING);

	/*
	 * 如果处于停止模式，则抛出错误。我们需要防止可能在关闭检查点
	 * 被写入时生成WAL的命令。为了安全起见，我们禁止所有新命令。
	 */
	if (MyWalSnd->state == WALSNDSTATE_STOPPING)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot execute new commands while WAL sender is in stopping mode")));

	/*
	 * CREATE_REPLICATION_SLOT ... LOGICAL 导出一个快照，直到下一个
	 * 命令到达。如果有任何旧内容，清除它。
	 */
	SnapBuildClearExportedSnapshot();

	CHECK_FOR_INTERRUPTS();

	/*
	 * 准备解析和执行命令。
	 */
	fc_cmd_context = AllocSetContextCreate(CurrentMemoryContext,
										"Replication command context",
										ALLOCSET_DEFAULT_SIZES);
	fc_old_context = MemoryContextSwitchTo(fc_cmd_context);

	replication_scanner_init(fc_cmd_string);

	/*
	 * 这是一条WalSender命令吗？
	 */
	if (!replication_scanner_is_replication_command())
	{
		/* 不；清理并退出。 */
		replication_scanner_finish();

		MemoryContextSwitchTo(fc_old_context);
		MemoryContextDelete(fc_cmd_context);

		/* XXX 这是检查的一个相当随机的地方 */
		if (MyDatabaseId == InvalidOid)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("cannot execute SQL commands in WAL sender for physical replication")));

		/* 告诉调用者这不是一个 WalSender 命令。 */
		return false;
	}

	/*
	 * 看起来是一个 WalSender 命令，所以解析它。
	 */
	fc_parse_rc = replication_yyparse();
	if (fc_parse_rc != 0)
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg_internal("replication command parser returned %d",
								 fc_parse_rc)));
	replication_scanner_finish();

	fc_cmd_node = replication_parse_result;

	/*
	 * 向各种监控设施报告查询。为了这个目的，我们将
	 * 复制命令报告得像 SQL 命令一样。
	 */
	debug_query_string = fc_cmd_string;

	pgstat_report_activity(STATE_RUNNING, fc_cmd_string);

	/*
	 * 如果 log_replication_commands 启用，则记录复制命令。即使
	 * 当它被禁用时，也要以 DEBUG1 级别记录命令以保持向后兼容性。
	 */
	ereport(log_replication_commands ? LOG : DEBUG1,
			(errmsg("received replication command: %s", fc_cmd_string)));

	/*
	 * 不允许在中止的事务块中执行复制命令。
	 */
	if (IsAbortedTransactionBlockState())
		ereport(ERROR,
				(errcode(ERRCODE_IN_FAILED_SQL_TRANSACTION),
				 errmsg("current transaction is aborted, "
						"commands ignored until end of transaction block")));

	CHECK_FOR_INTERRUPTS();

	/*
	 * 分配将用于每个传出和传入消息的缓冲区。我们每个命令
	 * 只做一次以减少 palloc 的开销。
	 */
	initStringInfo(&output_message);
	initStringInfo(&reply_message);
	initStringInfo(&tmpbuf);

	switch (fc_cmd_node->type)
	{
		case T_IdentifySystemCmd:
			fc_cmdtag = "IDENTIFY_SYSTEM";
			set_ps_display(fc_cmdtag);
			fc_IdentifySystem();
			EndReplicationCommand(fc_cmdtag);
			break;

		case T_ReadReplicationSlotCmd:
			fc_cmdtag = "READ_REPLICATION_SLOT";
			set_ps_display(fc_cmdtag);
			fc_ReadReplicationSlot((ReadReplicationSlotCmd *) fc_cmd_node);
			EndReplicationCommand(fc_cmdtag);
			break;

		case T_BaseBackupCmd:
			fc_cmdtag = "BASE_BACKUP";
			set_ps_display(fc_cmdtag);
			PreventInTransactionBlock(true, fc_cmdtag);
			SendBaseBackup((BaseBackupCmd *) fc_cmd_node);
			EndReplicationCommand(fc_cmdtag);
			break;

		case T_CreateReplicationSlotCmd:
			fc_cmdtag = "CREATE_REPLICATION_SLOT";
			set_ps_display(fc_cmdtag);
			fc_CreateReplicationSlot((CreateReplicationSlotCmd *) fc_cmd_node);
			EndReplicationCommand(fc_cmdtag);
			break;

		case T_DropReplicationSlotCmd:
			fc_cmdtag = "DROP_REPLICATION_SLOT";
			set_ps_display(fc_cmdtag);
			fc_DropReplicationSlot((DropReplicationSlotCmd *) fc_cmd_node);
			EndReplicationCommand(fc_cmdtag);
			break;

		case T_StartReplicationCmd:
			{
				StartReplicationCmd *fc_cmd = (StartReplicationCmd *) fc_cmd_node;

				fc_cmdtag = "START_REPLICATION";
				set_ps_display(fc_cmdtag);
				PreventInTransactionBlock(true, fc_cmdtag);

				if (fc_cmd->kind == REPLICATION_KIND_PHYSICAL)
					fc_StartReplication(fc_cmd);
				else
					fc_StartLogicalReplication(fc_cmd);

				/* 重复，但根据 libpqrcv_endstreaming 是必要的 */
				EndReplicationCommand(fc_cmdtag);

				Assert(xlogreader != NULL);
				break;
			}

		case T_TimeLineHistoryCmd:
			fc_cmdtag = "TIMELINE_HISTORY";
			set_ps_display(fc_cmdtag);
			PreventInTransactionBlock(true, fc_cmdtag);
			fc_SendTimeLineHistory((TimeLineHistoryCmd *) fc_cmd_node);
			EndReplicationCommand(fc_cmdtag);
			break;

		case T_VariableShowStmt:
			{
				DestReceiver *fc_dest = CreateDestReceiver(DestRemoteSimple);
				VariableShowStmt *fc_n = (VariableShowStmt *) fc_cmd_node;

				fc_cmdtag = "SHOW";
				set_ps_display(fc_cmdtag);

				/* syscache 访问需要事务环境 */
				StartTransactionCommand();
				GetPGVariable(fc_n->name, fc_dest);
				CommitTransactionCommand();
				EndReplicationCommand(fc_cmdtag);
			}
			break;

		default:
			elog(ERROR, "unrecognized replication command node tag: %u",
				 fc_cmd_node->type);
	}

	/* 完成 */
	MemoryContextSwitchTo(fc_old_context);
	MemoryContextDelete(fc_cmd_context);

	/*
	 * 我们不需要更新 ps 显示或 pg_stat_activity，因为 PostgresMain
	 * 将重置它们为“空闲”。但我们必须重置 debug_query_string 以
	 * 确保它不会成为悬挂指针。
	 */
	debug_query_string = NULL;

	return true;
}

/*
 * 在流式传输中处理任何传入消息。还检查远程端是否已关闭连接。
 */
static void fc_ProcessRepliesIfAny(void)
{
	unsigned char fc_firstchar;
	int			fc_maxmsglen;
	int			fc_r;
	bool		fc_received = false;

	last_processing = GetCurrentTimestamp();

	/*
	 * 如果我们已经从前端接收到 CopyDone，则任何后续消息都是
	 * 新命令的开始，应在主处理循环中处理。
	 */
	while (!streamingDoneReceiving)
	{
		pq_startmsgread();
		fc_r = pq_getbyte_if_available(&fc_firstchar);
		if (fc_r < 0)
		{
			/* 意外错误或 EOF */
			ereport(COMMERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("unexpected EOF on standby connection")));
			proc_exit(0);
		}
		if (fc_r == 0)
		{
			/* 没有可用数据，无法阻塞 */
			pq_endmsgread();
			break;
		}

		/* 验证消息类型并设置数据包大小限制 */
		switch (fc_firstchar)
		{
			case 'd':
				fc_maxmsglen = PQ_LARGE_MESSAGE_LIMIT;
				break;
			case 'c':
			case 'X':
				fc_maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
				break;
			default:
				ereport(FATAL,
						(errcode(ERRCODE_PROTOCOL_VIOLATION),
						 errmsg("invalid standby message type \"%c\"",
								fc_firstchar)));
				fc_maxmsglen = 0;	/* 保持编译器安静 */
				break;
		}

		/* 读取消息内容 */
		resetStringInfo(&reply_message);
		if (pq_getmessage(&reply_message, fc_maxmsglen))
		{
			ereport(COMMERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("unexpected EOF on standby connection")));
			proc_exit(0);
		}

		/* ...并处理它 */
		switch (fc_firstchar)
		{
				/*
				 * 'd' 表示包装在 CopyData 数据包中的备用回复。
				 */
			case 'd':
				fc_ProcessStandbyMessage();
				fc_received = true;
				break;

				/*
				 * CopyDone 表示备用请求结束流式传输。
				 * 如果我们之前没有发送过，回复 CopyDone。
				 */
			case 'c':
				if (!streamingDoneSending)
				{
					pq_putmessage_noblock('c', NULL, 0);
					streamingDoneSending = true;
				}

				streamingDoneReceiving = true;
				fc_received = true;
				break;

				/*
				 * 'X' 表示备用正在关闭套接字。
				 */
			case 'X':
				proc_exit(0);

			default:
				Assert(false);	/* NOT REACHED */
		}
	}

	/*
	 * 如果我们已经收到至少一个回复，则保存最后的回复时间戳。
	 */
	if (fc_received)
	{
		last_reply_timestamp = last_processing;
		waiting_for_ping_response = false;
	}
}

/*
 * 处理从备用接收到的状态更新消息。
 */
static void fc_ProcessStandbyMessage(void)
{
	char		fc_msgtype;

	/*
	 * 从第一个字节检查消息类型。
	 */
	fc_msgtype = pq_getmsgbyte(&reply_message);

	switch (fc_msgtype)
	{
		case 'r':
			fc_ProcessStandbyReplyMessage();
			break;

		case 'h':
			fc_ProcessStandbyHSFeedbackMessage();
			break;

		default:
			ereport(COMMERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("unexpected message type \"%c\"", fc_msgtype)));
			proc_exit(0);
	}
}

/*
 * 记住 walreceiver 刚确认收到 lsn `lsn`。
 */
static void fc_PhysicalConfirmReceivedLocation(XLogRecPtr fc_lsn)
{
	bool		fc_changed = false;
	ReplicationSlot *fc_slot = MyReplicationSlot;

	Assert(fc_lsn != InvalidXLogRecPtr);
	SpinLockAcquire(&fc_slot->mutex);
	if (fc_slot->data.restart_lsn != fc_lsn)
	{
		fc_changed = true;
		fc_slot->data.restart_lsn = fc_lsn;
	}
	SpinLockRelease(&fc_slot->mutex);

	if (fc_changed)
	{
		ReplicationSlotMarkDirty();
		ReplicationSlotsComputeRequiredLSN();
	}

	/*
	 * 有人可能会争辩说，槽应该立即保存到磁盘，但那样会
	 * 浪费能源 - 最糟糕的丢失信息会在统计视图中给我们提供
	 * 错误的信息 - 我们只会在删除文件时更加保守。
	 */
}

/*
 * 从备用发送的常规回复，告知备用服务器上的 WAL 位置。
 */
static void fc_ProcessStandbyReplyMessage(void)
{
	XLogRecPtr	fc_writePtr,
				fc_flushPtr,
				fc_applyPtr;
	bool		fc_replyRequested;
	TimeOffset	fc_writeLag,
				fc_flushLag,
				fc_applyLag;
	bool		fc_clearLagTimes;
	TimestampTz fc_now;
	TimestampTz fc_replyTime;

	static bool fc_fullyAppliedLastTime = false;

	/* 调用者已经消费了 msgtype 字节 */
	fc_writePtr = pq_getmsgint64(&reply_message);
	fc_flushPtr = pq_getmsgint64(&reply_message);
	fc_applyPtr = pq_getmsgint64(&reply_message);
	fc_replyTime = pq_getmsgint64(&reply_message);
	fc_replyRequested = pq_getmsgbyte(&reply_message);

	if (message_level_is_interesting(DEBUG2))
	{
		char	   *fc_replyTimeStr;

		/* 复制因为timestamptz_to_str返回一个静态缓冲区 */
		fc_replyTimeStr = pstrdup(timestamptz_to_str(fc_replyTime));

		elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X%s reply_time %s",
			 LSN_FORMAT_ARGS(fc_writePtr),
			 LSN_FORMAT_ARGS(fc_flushPtr),
			 LSN_FORMAT_ARGS(fc_applyPtr),
			 fc_replyRequested ? " (reply requested)" : "",
			 fc_replyTimeStr);

		pfree(fc_replyTimeStr);
	}

	/* 查看我们是否可以计算这些位置的往返延迟。 */
	fc_now = GetCurrentTimestamp();
	fc_writeLag = fc_LagTrackerRead(SYNC_REP_WAIT_WRITE, fc_writePtr, fc_now);
	fc_flushLag = fc_LagTrackerRead(SYNC_REP_WAIT_FLUSH, fc_flushPtr, fc_now);
	fc_applyLag = fc_LagTrackerRead(SYNC_REP_WAIT_APPLY, fc_applyPtr, fc_now);

	/*
	 * 如果备用报告它已在两个连续的回复消息中完全重放了 WAL，
	 * 则第二条此类消息必须是由于 wal_receiver_status_interval 在
	 * 备用上过期而导致的。这是忘记上次写入/刷新/应用 WAL 记录时
	 * 测量的延迟时间的方便时机，以避免在更多 WAL 流量到来之前
	 * 显示过时的延迟数据。
	 */
	fc_clearLagTimes = false;
	if (fc_applyPtr == sentPtr)
	{
		if (fc_fullyAppliedLastTime)
			fc_clearLagTimes = true;
		fc_fullyAppliedLastTime = true;
	}
	else
		fc_fullyAppliedLastTime = false;

	/* 如果备用请求回复，则发送回复。 */
	if (fc_replyRequested)
		fc_WalSndKeepalive(false, InvalidXLogRecPtr);

	/*
	 * 根据来自备用的回复数据更新此 WalSender 进程的共享状态。
	 */
	{
		WalSnd	   *fc_walsnd = MyWalSnd;

		SpinLockAcquire(&fc_walsnd->mutex);
		fc_walsnd->write = fc_writePtr;
		fc_walsnd->flush = fc_flushPtr;
		fc_walsnd->apply = fc_applyPtr;
		if (fc_writeLag != -1 || fc_clearLagTimes)
			fc_walsnd->writeLag = fc_writeLag;
		if (fc_flushLag != -1 || fc_clearLagTimes)
			fc_walsnd->flushLag = fc_flushLag;
		if (fc_applyLag != -1 || fc_clearLagTimes)
			fc_walsnd->applyLag = fc_applyLag;
		fc_walsnd->replyTime = fc_replyTime;
		SpinLockRelease(&fc_walsnd->mutex);
	}

	if (!am_cascading_walsender)
		SyncRepReleaseWaiters();

	/*
	 * 当客户端确认刷新时，推进我们的本地 xmin 地平线。
	 */
	if (MyReplicationSlot && fc_flushPtr != InvalidXLogRecPtr)
	{
		if (SlotIsLogical(MyReplicationSlot))
			LogicalConfirmReceivedLocation(fc_flushPtr);
		else
			fc_PhysicalConfirmReceivedLocation(fc_flushPtr);
	}
}

/* 如果需要，计算新的复制槽 xmin 地平线 */
static void fc_PhysicalReplicationSlotNewXmin(TransactionId fc_feedbackXmin, TransactionId fc_feedbackCatalogXmin)
{
	bool		fc_changed = false;
	ReplicationSlot *fc_slot = MyReplicationSlot;

	SpinLockAcquire(&fc_slot->mutex);
	MyProc->xmin = InvalidTransactionId;

	/*
 * 对于物理复制，我们不需要xmin和effective_xmin提供的互锁，因为遗漏增加的后果仅限于查询取消，因此同时设置两个值。
 */
	if (!TransactionIdIsNormal(fc_slot->data.xmin) ||
		!TransactionIdIsNormal(fc_feedbackXmin) ||
		TransactionIdPrecedes(fc_slot->data.xmin, fc_feedbackXmin))
	{
		fc_changed = true;
		fc_slot->data.xmin = fc_feedbackXmin;
		fc_slot->effective_xmin = fc_feedbackXmin;
	}
	if (!TransactionIdIsNormal(fc_slot->data.catalog_xmin) ||
		!TransactionIdIsNormal(fc_feedbackCatalogXmin) ||
		TransactionIdPrecedes(fc_slot->data.catalog_xmin, fc_feedbackCatalogXmin))
	{
		fc_changed = true;
		fc_slot->data.catalog_xmin = fc_feedbackCatalogXmin;
		fc_slot->effective_catalog_xmin = fc_feedbackCatalogXmin;
	}
	SpinLockRelease(&fc_slot->mutex);

	if (fc_changed)
	{
		ReplicationSlotMarkDirty();
		ReplicationSlotsComputeRequiredXmin(false);
	}
}

/*
 * 检查提供的xmin/epoch是否合理，即不在未来，并且不会过于久远以至于已经循环。
 *
 * nextXid的epoch应该与备用相同，或者如果计数器已经循环，则应比备用大一个。
 *
 * 此检查不关心这些xids是否存在clog。
 */
static bool fc_TransactionIdInRecentPast(TransactionId fc_xid, uint32 fc_epoch)
{
	FullTransactionId fc_nextFullXid;
	TransactionId fc_nextXid;
	uint32		fc_nextEpoch;

	fc_nextFullXid = ReadNextFullTransactionId();
	fc_nextXid = XidFromFullTransactionId(fc_nextFullXid);
	fc_nextEpoch = EpochFromFullTransactionId(fc_nextFullXid);

	if (fc_xid <= fc_nextXid)
	{
		if (fc_epoch != fc_nextEpoch)
			return false;
	}
	else
	{
		if (fc_epoch + 1 != fc_nextEpoch)
			return false;
	}

	if (!TransactionIdPrecedesOrEquals(fc_xid, fc_nextXid))
		return false;			/* epoch正常，但它已经循环 */

	return true;
}

/*
 * 热备用反馈
 */
static void fc_ProcessStandbyHSFeedbackMessage(void)
{
	TransactionId fc_feedbackXmin;
	uint32		fc_feedbackEpoch;
	TransactionId fc_feedbackCatalogXmin;
	uint32		fc_feedbackCatalogEpoch;
	TimestampTz fc_replyTime;

	/*
 * 解读回复消息。调用者已经消耗了msgtype字节。请参见walreceiver.c中的XLogWalRcvSendHSFeedback()以了解此消息的创建。
 */
	fc_replyTime = pq_getmsgint64(&reply_message);
	fc_feedbackXmin = pq_getmsgint(&reply_message, 4);
	fc_feedbackEpoch = pq_getmsgint(&reply_message, 4);
	fc_feedbackCatalogXmin = pq_getmsgint(&reply_message, 4);
	fc_feedbackCatalogEpoch = pq_getmsgint(&reply_message, 4);

	if (message_level_is_interesting(DEBUG2))
	{
		char	   *fc_replyTimeStr;

		/* 复制因为timestamptz_to_str返回一个静态缓冲区 */
		fc_replyTimeStr = pstrdup(timestamptz_to_str(fc_replyTime));

		elog(DEBUG2, "hot standby feedback xmin %u epoch %u, catalog_xmin %u epoch %u reply_time %s",
			 fc_feedbackXmin,
			 fc_feedbackEpoch,
			 fc_feedbackCatalogXmin,
			 fc_feedbackCatalogEpoch,
			 fc_replyTimeStr);

		pfree(fc_replyTimeStr);
	}

	/*
	 * 根据来自备用的回复数据更新此 WalSender 进程的共享状态。
	 */
	{
		WalSnd	   *fc_walsnd = MyWalSnd;

		SpinLockAcquire(&fc_walsnd->mutex);
		fc_walsnd->replyTime = fc_replyTime;
		SpinLockRelease(&fc_walsnd->mutex);
	}

	/*
 * 如果反馈消息值无效，则取消WalSender的xmins。这发生在下游关闭hot_standby_feedback时。
 */
	if (!TransactionIdIsNormal(fc_feedbackXmin)
		&& !TransactionIdIsNormal(fc_feedbackCatalogXmin))
	{
		MyProc->xmin = InvalidTransactionId;
		if (MyReplicationSlot != NULL)
			fc_PhysicalReplicationSlotNewXmin(fc_feedbackXmin, fc_feedbackCatalogXmin);
		return;
	}

	/*
 * 检查提供的xmin/epoch是否合理，即不在未来，并且不会过于久远以至于已经循环。如果不是，则忽略。
 */
	if (TransactionIdIsNormal(fc_feedbackXmin) &&
		!fc_TransactionIdInRecentPast(fc_feedbackXmin, fc_feedbackEpoch))
		return;

	if (TransactionIdIsNormal(fc_feedbackCatalogXmin) &&
		!fc_TransactionIdInRecentPast(fc_feedbackCatalogXmin, fc_feedbackCatalogEpoch))
		return;

	/*
 * 将WalSender的xmin设置为备用请求的xmin，以便xmin将被GetSnapshotData()/ComputeXidHorizons()考虑。这样可以延迟删除死行，从而防止备用服务器上发生清理冲突。
 *
 * 这里存在一个小的竞争条件窗口：虽然我们刚刚检查了feedbackXmin早于nextXid，但nextXid可能在我们获取它和应用下面的xmin之间得到了推进，可能推进得足够远以使feedbackXmin循环。在这种情况下，我们在这里设置的xmin将“在未来”，并且不会产生影响。担心这个没有意义，因为无论如何已经太晚了，也无法保存所需数据。假设备用给我们发送的是一个递增的xmin序列，则这种情况只会发生在第一次回复周期，否则我们的xmin会防止nextXid推进得如此之远。
 *
 * 我们在这里不费心获取ProcArrayLock。设置xmin字段被认为是原子的，并且实际上没有必要防止并发的地平线确定。（如果我们正在向前移动我们的xmin，这显然是安全的，而如果我们向后移动它，好吧，数据已经处于风险中，因为VACUUM可能已经确定了地平线。）
 *
 * 如果我们正在使用复制槽，我们通过该槽保留xmin，否则通过walsender的PGPROC条目保留。当不使用插槽时，我们只能分别跟踪目录xmin，因此我们存储两个提供数值中的最小值。
 *
 * XXX：使持久性可配置可能是有意义的？
 */
	if (MyReplicationSlot != NULL)	/* XXX: 持久性可配置？ */
		fc_PhysicalReplicationSlotNewXmin(fc_feedbackXmin, fc_feedbackCatalogXmin);
	else
	{
		if (TransactionIdIsNormal(fc_feedbackCatalogXmin)
			&& TransactionIdPrecedes(fc_feedbackCatalogXmin, fc_feedbackXmin))
			MyProc->xmin = fc_feedbackCatalogXmin;
		else
			MyProc->xmin = fc_feedbackXmin;
	}
}

/*
 * 计算发送/接收循环应休眠多久。
 *
 * 如果wal_sender_timeout已启用，我们希望及时唤醒以发送保持活动消息，并在达到wal_sender_timeout时中止连接。
 */
static long fc_WalSndComputeSleeptime(TimestampTz fc_now)
{
	long		fc_sleeptime = 10000;	/* 10 s */

	if (wal_sender_timeout > 0 && last_reply_timestamp > 0)
	{
		TimestampTz fc_wakeup_time;

		/*
 * 最迟在达到wal_sender_timeout时停止休眠。
 */
		fc_wakeup_time = TimestampTzPlusMilliseconds(last_reply_timestamp,
												  wal_sender_timeout);

		/*
 * 如果尚未发送ping，则在该发送ping的时间唤醒。WalSndKeepaliveIfNecessary()希望在超过一半的超时后发送保持活动消息，而没有响应。
 */
		if (!waiting_for_ping_response)
			fc_wakeup_time = TimestampTzPlusMilliseconds(last_reply_timestamp,
													  wal_sender_timeout / 2);

		/* 计算直到唤醒的相对时间。 */
		fc_sleeptime = TimestampDifferenceMilliseconds(fc_now, fc_wakeup_time);
	}

	return fc_sleeptime;
}

/*
 * 检查在wal_sender_timeout内客户是否有响应，如果没有则关闭。使用last_processing作为参考点可以避免将服务器侧的暂停算作客户端的。 然而，较长的服务器侧暂停可能会使WalSndKeepaliveIfNecessary()将last_processing推迟到超过wal_sender_timeout。如果发生这种情况，客户端必须几乎立即回复以避免超时。这通常不会影响默认配置，在该配置下，客户端自发地每个standby_message_timeout = wal_sender_timeout/6 = 10s发送一条消息。我们可以通过在保持活动的wal_sender_timeout/2之前识别超时失效来消除该问题。
 */
static void fc_WalSndCheckTimeOut(void)
{
	TimestampTz fc_timeout;

	/* 如果我们正在做的事情不需要超时，则不退出 */
	if (last_reply_timestamp <= 0)
		return;

	fc_timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
										  wal_sender_timeout);

	if (wal_sender_timeout > 0 && last_processing >= fc_timeout)
	{
		/*
 * 由于通常情况下复制超时的过期意味着通信问题，我们不会将错误消息发送给备用。
 */
		ereport(COMMERROR,
				(errmsg("terminating walsender process due to replication timeout")));

		fc_WalSndShutdown();
	}
}

/* 流式传输WAL的walsender进程的主循环。 */
static void fc_WalSndLoop(WalSndSendDataCallback fc_send_data)
{
	/*
 * 初始化最后的回复时间戳。这使得从现在开始可以进行超时处理。
 */
	last_reply_timestamp = GetCurrentTimestamp();
	waiting_for_ping_response = false;

	/*
 * 循环直到我们到达此时间线的末尾或客户端请求停止流媒体。
 */
	for (;;)
	{
		/* 清除任何已经等待的唤醒 */
		ResetLatch(MyLatch);

		CHECK_FOR_INTERRUPTS();

		/* 处理最近收到的任何请求或信号 */
		if (ConfigReloadPending)
		{
			ConfigReloadPending = false;
			ProcessConfigFile(PGC_SIGHUP);
#ifdef FDD
			SyncRepInitConfig(MyReplicationSlot != NULL ? MyReplicationSlot->data.failover : false);
#else
			SyncRepInitConfig();
#endif
		}

		/* 检查来自客户端的输入 */
		fc_ProcessRepliesIfAny();

		/*
 * 如果我们已从客户端收到CopyDone，自己发送了CopyDone，并且输出缓冲区为空，那么该退出流媒体的时机已到。
 */
		if (streamingDoneReceiving && streamingDoneSending &&
			!pq_is_send_pending())
			break;

		
/*
		 * 如果我们在输出缓冲区没有任何待处理的数据，尝试发送
		 * 更多的数据。如果有数据，我们不会再次调用 send_data
		 * 直到我们刷新它……但我们最好假设我们并没有
		 * 跟上进度。
		 */
		if (!pq_is_send_pending())
			fc_send_data();
		else
			WalSndCaughtUp = false;

		/* 尝试将待处理输出刷新到客户端 */
		if (pq_flush_if_writable() != 0)
			fc_WalSndShutdown();

		/* 如果现在没有剩余的数据要发送…… */
		if (WalSndCaughtUp && !pq_is_send_pending())
		{
			/*
			 * 如果我们处于追赶状态，转向流式传输。这是一个
			 * 重要的状态变化，用户需要了解，因为在此之前
			 * 如果主服务器崩溃并且我们需要切换到备用服务器时，
			 * 可能会发生数据丢失。这个状态变化对于
			 * 同步复制也很重要，因为在那一点开始等待的提交
			 * 可能会等待一段时间。
			 */
			if (MyWalSnd->state == WALSNDSTATE_CATCHUP)
			{
				ereport(DEBUG1,
						(errmsg_internal("\"%s\" has now caught up with upstream server",
										 application_name)));
				WalSndSetState(WALSNDSTATE_STREAMING);
			}

			/*
			 * 当收到 SIGUSR2 时，我们会发送所有未处理的日志
			 * 直到关闭检查点记录（即最新记录），等待
			 * 它们被复制到备用服务器，然后退出。这可能是在关闭时
			 * 正常终止，或者是提升，walsender
			 * 不确定是哪种情况。
			 */
			if (got_SIGUSR2)
				fc_WalSndDone(fc_send_data);
		}

		/* 检查复制超时。 */
		fc_WalSndCheckTimeOut();

		/* 如果时间到了则发送保持连接消息 */
		fc_WalSndKeepaliveIfNecessary();

		/*
		 * 如果有未发送的数据则阻塞。XXX 对于逻辑复制，让
		 * WalSndWaitForWal() 处理任何其他阻塞；空闲接收器需要
		 * 它的额外操作。对于物理复制，如果已经赶上，也要阻塞；
		 * 它的 send_data 不会阻塞。
		 */
		if ((WalSndCaughtUp && fc_send_data != fc_XLogSendLogical &&
			 !streamingDoneSending) ||
			pq_is_send_pending())
		{
			long		fc_sleeptime;
			int			fc_wakeEvents;

			if (!streamingDoneReceiving)
				fc_wakeEvents = WL_SOCKET_READABLE;
			else
				fc_wakeEvents = 0;

			/*
			 * 使用新鲜的时间戳，而不是 last_processing，以减少
			 * 在发送保持活动消息之前达到 wal_sender_timeout 的机会。
			 */
			fc_sleeptime = fc_WalSndComputeSleeptime(GetCurrentTimestamp());

			if (pq_is_send_pending())
				fc_wakeEvents |= WL_SOCKET_WRITEABLE;

			/* 睡眠直到发生某事或我们超时 */
			fc_WalSndWait(fc_wakeEvents, fc_sleeptime, WAIT_EVENT_WAL_SENDER_MAIN);
		}
	}
}

/* 为这个 walsender 进程初始化一个每个 walsender 的数据结构 */
static void fc_InitWalSenderSlot(void)
{
	int			fc_i;

	/*
	 * WalSndCtl 应该已经设置好（我们通过 fork() 或
	 * EXEC_BACKEND 机制从主进程继承这个）。
	 */
	Assert(WalSndCtl != NULL);
	Assert(MyWalSnd == NULL);

	/*
	 * 找到一个空闲的 walsender 插槽并保留它。这必须不会失败，因为
	 * 在 InitProcess() 中已经检查过空闲 WAL 发送器。
	 */
	for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
	{
		WalSnd	   *fc_walsnd = &WalSndCtl->walsnds[fc_i];

		SpinLockAcquire(&fc_walsnd->mutex);

		if (fc_walsnd->pid != 0)
		{
			SpinLockRelease(&fc_walsnd->mutex);
			continue;
		}
		else
		{
			/*
			 * 找到一个空闲的槽。为我们预留它。
			 */
			fc_walsnd->pid = MyProcPid;
			fc_walsnd->state = WALSNDSTATE_STARTUP;
			fc_walsnd->sentPtr = InvalidXLogRecPtr;
			fc_walsnd->needreload = false;
			fc_walsnd->write = InvalidXLogRecPtr;
			fc_walsnd->flush = InvalidXLogRecPtr;
			fc_walsnd->apply = InvalidXLogRecPtr;
			fc_walsnd->writeLag = -1;
			fc_walsnd->flushLag = -1;
			fc_walsnd->applyLag = -1;
			fc_walsnd->sync_standby_priority = 0;
			fc_walsnd->latch = &MyProc->procLatch;
			fc_walsnd->replyTime = 0;
			SpinLockRelease(&fc_walsnd->mutex);
			/* 不再需要锁 */
			MyWalSnd = (WalSnd *) fc_walsnd;

			break;
		}
	}

	Assert(MyWalSnd != NULL);

	/* 安排在 walsender 退出时进行清理 */
	on_shmem_exit(fc_WalSndKill, 0);
}

/* 销毁此 walsender 进程的每个 walsender 数据结构 */
static void fc_WalSndKill(int fc_code, Datum fc_arg)
{
	WalSnd	   *fc_walsnd = MyWalSnd;

	Assert(fc_walsnd != NULL);

	MyWalSnd = NULL;

	SpinLockAcquire(&fc_walsnd->mutex);
	/* 在持有自旋锁的同时清除闩锁，以便可以安全读取 */
	fc_walsnd->latch = NULL;
	/* 将 WalSnd 结构标记为不再使用 */
	fc_walsnd->pid = 0;
	SpinLockRelease(&fc_walsnd->mutex);
}

/* XLogReaderRoutine->segment_open 回调 */
static void fc_WalSndSegmentOpen(XLogReaderState *fc_state, XLogSegNo fc_nextSegNo,
				  TimeLineID *fc_tli_p)
{
	char		fc_path[MAXPGPATH];

	/*-------
	 * 当从历史时间线读取时，如果在此段内有时间线切换，
	 * 则从属于新时间线的 WAL 段读取。
	 *
	 * 例如，假设这个服务器当前处于时间线 5，且我们正在流式传输时间线 4。
	 * 从时间线 4 到 5 的切换发生在 0/13002088。在 pg_wal 中，我们有这些文件：
	 *
	 * ...
	 * 000000040000000000000012
	 * 000000040000000000000013
	 * 000000050000000000000013
	 * 000000050000000000000014
	 * ...
	 *
	 * 在这种情况下，当请求发送来自时间线 4 的段 0x13 的 WAL 时，
	 * 我们从文件 000000050000000000000013 读取 WAL。归档恢复更倾向于
	 * 使用较新时间线的文件，因此如果该段是从本服务器的归档恢复的，
	 * 属于旧时间线的文件 000000040000000000000013 可能不存在。
	 * 它们的内容在切换点之前是相等的，因为在时间线切换时，
	 * 旧段的已使用部分被复制到新文件中。-------
	 */
	*fc_tli_p = sendTimeLine;
	if (sendTimeLineIsHistoric)
	{
		XLogSegNo	fc_endSegNo;

		XLByteToSeg(sendTimeLineValidUpto, fc_endSegNo, fc_state->segcxt.ws_segsize);
		if (fc_nextSegNo == fc_endSegNo)
			*fc_tli_p = sendTimeLineNextTLI;
	}

	XLogFilePath(fc_path, *fc_tli_p, fc_nextSegNo, fc_state->segcxt.ws_segsize);
	fc_state->seg.ws_file = BasicOpenFile(fc_path, O_RDONLY | PG_BINARY);
	if (fc_state->seg.ws_file >= 0)
		return;

	/*
	 * 如果找不到文件，假设是因为备用请求了一个过旧的
	 * WAL 段，该段已被删除或回收。
	 */
	if (errno == ENOENT)
	{
		char		fc_xlogfname[MAXFNAMELEN];
		int			fc_save_errno = errno;

		XLogFileName(fc_xlogfname, *fc_tli_p, fc_nextSegNo, wal_segment_size);
		errno = fc_save_errno;
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("requested WAL segment %s has already been removed",
						fc_xlogfname)));
	}
	else
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not open file \"%s\": %m",
						fc_path)));
}

/*
 * 以正常的物理/存储形式发送 WAL。
 *
 * 读取最多 MAX_SEND_SIZE 字节的已刷新到磁盘但尚未发送到客户端的 WAL，
 * 并将其缓冲在 libpq 输出缓冲区中。
 *
 * 如果没有未发送的 WAL 剩余，WalSndCaughtUp 被设置为 true，
 * 否则 WalSndCaughtUp 被设置为 false。
 */
static void fc_XLogSendPhysical(void)
{
	XLogRecPtr	fc_SendRqstPtr;
	XLogRecPtr	fc_startptr;
	XLogRecPtr	fc_endptr;
	Size		fc_nbytes;
	XLogSegNo	fc_segno;
	WALReadError fc_errinfo;

	/* 如果请求，切换 WAL 发送器到停止状态。 */
	if (got_STOPPING)
		WalSndSetState(WALSNDSTATE_STOPPING);

	if (streamingDoneSending)
	{
		WalSndCaughtUp = true;
		return;
	}

	/* 计算我们可以安全发送 WAL 的距离。 */
	if (sendTimeLineIsHistoric)
	{
		/*
		 * 流式传输这个服务器历史中的旧时间线，但
		 * 不是我们当前插入或重放的时间线。可以流式传输
		 * 直到我们切换关闭那个时间线的点。
		 */
		fc_SendRqstPtr = sendTimeLineValidUpto;
	}
	else if (am_cascading_walsender)
	{
		TimeLineID	fc_SendRqstTLI;

		/*
		 * 在备用上流式传输最新时间线。
		 *
		 * 尝试发送所有已重放的 WAL，以便我们知道它是有效的。
		 * 如果我们通过流复制接收 WAL，发送任何已接收
		 * 但未重放的 WAL 也是可以的。
		 *
		 * 我们正在恢复的时间线可能会变化，或者我们可能被提升。
		 * 在任何情况下，当前时间线都变成历史时间线。
		 * 我们需要检测到这一点，以便不尝试流式传输过
		 * 切换到另一条时间线的点。在计算 FlushPtr 后检查是否提升或切换时间线
		 * ，以避免竞态条件：如果时间线在我们检查
		 * 之前仍为当前时间线后变为历史时间线，它仍然应该可以流式传输
		 * 到在变为历史时间线之前计算的 FlushPtr。
		 */
		bool		fc_becameHistoric = false;

		fc_SendRqstPtr = fc_GetStandbyFlushRecPtr(&fc_SendRqstTLI);

		if (!RecoveryInProgress())
		{
			/* 我们已经被提升。 */
			fc_SendRqstTLI = GetWALInsertionTimeLine();
			am_cascading_walsender = false;
			fc_becameHistoric = true;
		}
		else
		{
			/*
			 * 仍然是级联备用。但我们发送的时间线是否
			 * 仍然是恢复正在恢复的时间线？
			 */
			if (sendTimeLine != fc_SendRqstTLI)
				fc_becameHistoric = true;
		}

		if (fc_becameHistoric)
		{
			/*
			 * 我们发送的时间线已变为历史时间线。
			 * 阅读新时间线的时间线历史文件，以查看
			 * 我们确切地在哪个位置从要发送的时间线分叉。
			 */
			List	   *fc_history;

			fc_history = readTimeLineHistory(fc_SendRqstTLI);
			sendTimeLineValidUpto = tliSwitchPoint(sendTimeLine, fc_history, &sendTimeLineNextTLI);

			Assert(sendTimeLine < sendTimeLineNextTLI);
			list_free_deep(fc_history);

			sendTimeLineIsHistoric = true;

			fc_SendRqstPtr = sendTimeLineValidUpto;
		}
	}
	else
	{
		/*
		 * 在主节点上流式传输当前时间线。
		 *
		 * 尝试发送所有已写出并已 fsync 到磁盘的数据。
		 * 鉴于 WALRead() 的当前实现，我们无法超出
		 * 已写出数据的范围。而且在任何情况下，
		 * 发送尚未安全写入主节点磁盘的 WAL 是不安全的：
		 * 如果主节点随后崩溃并重新启动，备用节点
		 * 不能应用在主节点上丢失的任何 WAL。
		 */
		fc_SendRqstPtr = GetFlushRecPtr(NULL);
	}

	
/*
	 * 记录当前系统时间作为此 WAL 位置写入时的近似时间
	 * 以便进行延迟追踪。
	 *
	 * 理论上，我们可以让 XLogFlush() 在共享内存中记录时间，
	 * 每当 WAL 被刷新时，我们可以在调用 GetFlushRecPtr() 时获得该时间
	 * 和 LSN（级联备用的情况也是如此），但是与其在热 WAL 路径中添加任何新代码，
	 * 不如在这里捕获时间。我们应该在 XLogFlush() 执行 WalSndWakeupProcessRequests()
	 * 后达到这一点，尽管这可能需要一些时间，我们读取 WAL 刷新指针并在这里几乎同时
	 * 获取时间，以便如果它仍在移动，我们将获得一个更晚的位置。
	 *
	 * 由于 LagTrackerWrite 在 LSN 没有推进时会忽略样本，
	 * 这为该 LSN 提供了 WAL 刷新时间的廉价近似。
	 *
	 * 注意，LSN 不一定是当前消息中包含数据的 LSN；
	 * 它是 WAL 的结束，可能更靠前。所有延迟追踪机制关心的
	 * 是找出该任意 LSN 最终被报告为已写入、已刷新和已应用的时间，
	 * 这样它就可以测量经过的时间。
	 */
	fc_LagTrackerWrite(fc_SendRqstPtr, GetCurrentTimestamp());

	/*
	 * 如果这是一个历史时间线，并且我们已经达到了分叉到下一个时间线的点，停止流式传输。
	 *
	 * 注意：我们可能已经发送了 WAL > sendTimeLineValidUpto。启动进程通常会回放
	 * 从主节点接收到的所有 WAL，随后提升，但是如果在 WAL 页边界
	 * 终止 WAL 流式传输，时间线的有效部分可能会在 WAL 记录的中间结束。
	 * 我们可能已经将该部分 WAL 记录的前半部分发送给级联备用，因此
	 * sentPtr > sendTimeLineValidUpto。这没关系；级联备用也无法回放部分 WAL 记录，
	 * 所以它仍然可以跟随我们的时间线切换。
	 */
	if (sendTimeLineIsHistoric && sendTimeLineValidUpto <= sentPtr)
	{
		/* 关闭当前文件。 */
		if (xlogreader->seg.ws_file >= 0)
			wal_segment_close(xlogreader);

		/* 发送 CopyDone */
		pq_putmessage_noblock('c', NULL, 0);
		streamingDoneSending = true;

		WalSndCaughtUp = true;

		elog(DEBUG1, "walsender reached end of timeline at %X/%X (sent up to %X/%X)",
			 LSN_FORMAT_ARGS(sendTimeLineValidUpto),
			 LSN_FORMAT_ARGS(sentPtr));
		return;
	}

	/* 我们还有工作要做吗？ */
	Assert(sentPtr <= fc_SendRqstPtr);
	if (fc_SendRqstPtr <= sentPtr)
	{
		WalSndCaughtUp = true;
		return;
	}

	/*
	 * 计算在一条消息中发送多少。如果没有超过
	 * MAX_SEND_SIZE 字节要发送，则发送所有内容。否则发送
	 * MAX_SEND_SIZE 字节，但环绕到日志文件或页面边界。
	 *
	 * 进行这个环绕不仅是出于性能原因。Walreceiver 依赖于
	 * 我们从不在两条消息中拆分 WAL 记录这一事实。
	 * 由于长 WAL 记录在页面边界被拆分为继续记录，
	 * 页面边界始终是一个安全的切割点。我们还假设
	 * SendRqstPtr 从不指向 WAL 记录的中间。
	 */
	fc_startptr = sentPtr;
	fc_endptr = fc_startptr;
	fc_endptr += MAX_SEND_SIZE;

	/* 如果我们超出了 SendRqstPtr，退回 */
	if (fc_SendRqstPtr <= fc_endptr)
	{
		fc_endptr = fc_SendRqstPtr;
		if (sendTimeLineIsHistoric)
			WalSndCaughtUp = false;
		else
			WalSndCaughtUp = true;
	}
	else
	{
		/* 向下环绕到页面边界。 */
		fc_endptr -= (fc_endptr % XLOG_BLCKSZ);
		WalSndCaughtUp = false;
	}

	fc_nbytes = fc_endptr - fc_startptr;
	Assert(fc_nbytes <= MAX_SEND_SIZE);

	/*
	 * 可以读取并发送该片段。
	 */
	resetStringInfo(&output_message);
	pq_sendbyte(&output_message, 'w');

	pq_sendint64(&output_message, fc_startptr);	/* dataStart */
	pq_sendint64(&output_message, fc_SendRqstPtr); /* walEnd */
	pq_sendint64(&output_message, 0);	/* sendtime，最后填写 */

	/*
	 * 直接将日志读入输出缓冲区，以避免额外的 memcpy
	 * 调用。
	 */
	enlargeStringInfo(&output_message, fc_nbytes);

retry:
	if (!WALRead(xlogreader,
				 &output_message.data[output_message.len],
				 fc_startptr,
				 fc_nbytes,
				 xlogreader->seg.ws_tli,	/* 传递当前 TLI，因为
											 * 只有 WalSndSegmentOpen 控制
											 * 是否需要新的 TLI。 */
				 &fc_errinfo))
		WALReadRaiseError(&fc_errinfo);

	/* 查看 logical_read_xlog_page()。 */
	XLByteToSeg(fc_startptr, fc_segno, xlogreader->segcxt.ws_segsize);
	CheckXLogRemoved(fc_segno, xlogreader->seg.ws_tli);

	/*
	 * 在恢复期间，当前打开的 WAL 文件可能会被从归档中检索的同名文件替换。所以我们始终需要在将内容读入缓冲区后检查我们读取的内容是否有效。如果无效，我们尝试再次打开并读取该文件。
	 */
	if (am_cascading_walsender)
	{
		WalSnd	   *fc_walsnd = MyWalSnd;
		bool		fc_reload;

		SpinLockAcquire(&fc_walsnd->mutex);
		fc_reload = fc_walsnd->needreload;
		fc_walsnd->needreload = false;
		SpinLockRelease(&fc_walsnd->mutex);

		if (fc_reload && xlogreader->seg.ws_file >= 0)
		{
			wal_segment_close(xlogreader);

			goto retry;
		}
	}

	output_message.len += fc_nbytes;
	output_message.data[output_message.len] = '\0';

	/*
	 * 最后填充发送时间戳，以确保尽可能晚地获取它。
	 */
	resetStringInfo(&tmpbuf);
	pq_sendint64(&tmpbuf, GetCurrentTimestamp());
	memcpy(&output_message.data[1 + sizeof(int64) + sizeof(int64)],
		   tmpbuf.data, sizeof(int64));

	pq_putmessage_noblock('d', output_message.data, output_message.len);

	sentPtr = fc_endptr;

	/* 更新共享内存状态 */
	{
		WalSnd	   *fc_walsnd = MyWalSnd;

		SpinLockAcquire(&fc_walsnd->mutex);
		fc_walsnd->sentPtr = sentPtr;
		SpinLockRelease(&fc_walsnd->mutex);
	}

	/* 在 PS 显示中报告 XLOG 流媒体的进度 */
	if (update_process_title)
	{
		char		fc_activitymsg[50];

		snprintf(fc_activitymsg, sizeof(fc_activitymsg), "streaming %X/%X",
				 LSN_FORMAT_ARGS(sentPtr));
		set_ps_display(fc_activitymsg);
	}
}

/*
 * 逐步流出逻辑解码的数据。
 */
static void fc_XLogSendLogical(void)
{
	XLogRecord *fc_record;
	char	   *fc_errm;

	/*
	 * 我们将使用当前的刷新点来确定我们是否已经赶上了。该变量是静态的，以便在调用之间进行缓存。缓存很有帮助，因为 GetFlushRecPtr() 需要获取一个竞争激烈的自旋锁。
	 */
	static XLogRecPtr fc_flushPtr = InvalidXLogRecPtr;

	/*
	 * 不知道我们是否已经赶上了。如果我们实际上在等待，我们会在 WalSndWaitForWal 中将 WalSndCaughtUp 设置为 true。当 XLogReadRecord() 必须停止读取但 WalSndWaitForWal 没有等待时 - 例如，当我们关闭时，我们也将其设置为 true。
	 */
	WalSndCaughtUp = false;

	fc_record = XLogReadRecord(logical_decoding_ctx->reader, &fc_errm);

	/* xlog 记录无效 */
	if (fc_errm != NULL)
		elog(ERROR, "could not find record while sending logically-decoded data: %s",
			 fc_errm);

	if (fc_record != NULL)
	{
		/*
		 * 注意，没有调用 LagTrackerWrite()，该函数由 WalSndUpdateProgress 处理，并由输出插件通过逻辑解码写入 api 调用。
		 */
		LogicalDecodingProcessRecord(logical_decoding_ctx, logical_decoding_ctx->reader);

		sentPtr = logical_decoding_ctx->reader->EndRecPtr;
	}

	/*
	 * 如果这是本会话中的第一次，初始化 flushPtr。否则，仅当 EndRecPtr 超过 flushPtr 时，我们才需要更新 flushPtr。
	 */
	if (fc_flushPtr == InvalidXLogRecPtr)
		fc_flushPtr = GetFlushRecPtr(NULL);
	else if (logical_decoding_ctx->reader->EndRecPtr >= fc_flushPtr)
		fc_flushPtr = GetFlushRecPtr(NULL);

	/* 如果 EndRecPtr 仍然超过我们的 flushPtr，则意味着我们赶上了。 */
	if (logical_decoding_ctx->reader->EndRecPtr >= fc_flushPtr)
		WalSndCaughtUp = true;

	/*
	 * 如果我们赶上了并被请求停止，则让 WalSndLoop() 以有序的方式终止连接，在写出所有待处理的数据后。
	 */
	if (WalSndCaughtUp && got_STOPPING)
		got_SIGUSR2 = true;

	/* 更新共享内存状态 */
	{
		WalSnd	   *fc_walsnd = MyWalSnd;

		SpinLockAcquire(&fc_walsnd->mutex);
		fc_walsnd->sentPtr = sentPtr;
		SpinLockRelease(&fc_walsnd->mutex);
	}
}

/*
 * 如果发送者已赶上，则关闭。
 *
 * 注意：此函数仅应在从主进程接收到关闭信号时调用。
 *
 * 请注意，如果我们确定仍有更多数据要发送，此函数将将控制权返回给调用者。
 */
static void fc_WalSndDone(WalSndSendDataCallback fc_send_data)
{
	XLogRecPtr	fc_replicatedPtr;

	/* ... 让我们确保我们已经赶上了 ... */
	fc_send_data();

	/*
	 * 要确定所有 WAL 是否已成功复制，请检查刷新位置是否有效，否则写入。像 pg_receivewal 这样的工具通常（除非处于同步模式）会返回无效的刷新位置。
	 */
	fc_replicatedPtr = XLogRecPtrIsInvalid(MyWalSnd->flush) ?
		MyWalSnd->write : MyWalSnd->flush;

	if (WalSndCaughtUp && sentPtr == fc_replicatedPtr &&
		!pq_is_send_pending())
	{
		QueryCompletion fc_qc;

		/* 通知备用服务器 XLOG 流媒体已完成 */
		SetQueryCompletion(&fc_qc, CMDTAG_COPY, 0);
		EndCommand(&fc_qc, DestRemote, false);
		pq_flush();

		proc_exit(0);
	}
	if (!waiting_for_ping_response)
		fc_WalSndKeepalive(true, InvalidXLogRecPtr);
}

/*
 * 返回已安全刷新到磁盘的 WAL 中的最新点，并可以发送到备用服务器。这应仅在恢复期间调用，即我们正在流式传输到级联备用服务器。
 *
 * 作为副作用，*tli 被更新为最后播放的 WAL 记录的 TLI。
 */
static XLogRecPtr fc_GetStandbyFlushRecPtr(TimeLineID *fc_tli)
{
	XLogRecPtr	fc_replayPtr;
	TimeLineID	fc_replayTLI;
	XLogRecPtr	fc_receivePtr;
	TimeLineID	fc_receiveTLI;
	XLogRecPtr	fc_result;

	/*
	 * 我们可以安全地发送已经重放的内容。如果 walreceiver 正在从同一时间线流式传输 WAL，我们可以发送它所流式传输但尚未重放的任何内容。
	 */

	fc_receivePtr = GetWalRcvFlushRecPtr(NULL, &fc_receiveTLI);
	fc_replayPtr = GetXLogReplayRecPtr(&fc_replayTLI);

	*fc_tli = fc_replayTLI;

	fc_result = fc_replayPtr;
	if (fc_receiveTLI == fc_replayTLI && fc_receivePtr > fc_replayPtr)
		fc_result = fc_receivePtr;

	return fc_result;
}

/*
 * 请求 walsenders 重新加载当前打开的 WAL 文件
 */
void WalSndRqstFileReload(void)
{
	int			fc_i;

	for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
	{
		WalSnd	   *fc_walsnd = &WalSndCtl->walsnds[fc_i];

		SpinLockAcquire(&fc_walsnd->mutex);
		if (fc_walsnd->pid == 0)
		{
			SpinLockRelease(&fc_walsnd->mutex);
			continue;
		}
		fc_walsnd->needreload = true;
		SpinLockRelease(&fc_walsnd->mutex);
	}
}

/*
 * 处理 PROCSIG_WALSND_INIT_STOPPING 信号。
 */
void HandleWalSndInitStopping(void)
{
	Assert(am_walsender);

	/*
	 * 如果复制尚未开始，像 SIGTERM 一样终止。如果复制处于活动状态，则仅设置一个标志并唤醒主循环。它将发送任何未完成的 WAL，等待其被复制到备用服务器，然后优雅地退出。
	 */
	if (!replication_active)
		kill(MyProcPid, SIGTERM);
	else
		got_STOPPING = true;
}

/*
 * SIGUSR2：设置标志以进行最后一个周期，然后关闭。此时，WAL 发送器应该已经切换到 WALSNDSTATE_STOPPING 状态。
 */
static void fc_WalSndLastCycleHandler(SIGNAL_ARGS)
{
	int			fc_save_errno = errno;

	got_SIGUSR2 = true;
	SetLatch(MyLatch);

	errno = fc_save_errno;
}

/* 设置信号处理程序 */
void WalSndSignals(void)
{
	/* 设置信号处理程序 */
	pqsignal(SIGHUP, SignalHandlerForConfigReload);
	pqsignal(SIGINT, StatementCancelHandler);	/* 查询取消 */
	pqsignal(SIGTERM, die);		/* 请求关闭 */
	/* SIGQUIT 处理程序已经由 InitPostmasterChild 设置 */
	InitializeTimeouts();		/* 建立 SIGALRM 处理程序 */
	pqsignal(SIGPIPE, SIG_IGN);
	pqsignal(SIGUSR1, procsignal_sigusr1_handler);
	pqsignal(SIGUSR2, fc_WalSndLastCycleHandler);	/* 请求最后一个周期并
												 * 关闭 */

	/* 重置一些由 postmaster 接受但在这里不接受的信号 */
	pqsignal(SIGCHLD, SIG_DFL);
}

/* 报告 WalSndShmemInit 所需的共享内存空间 */
Size WalSndShmemSize(void)
{
	Size		fc_size = 0;

	fc_size = offsetof(WalSndCtlData, walsnds);
	fc_size = add_size(fc_size, mul_size(max_wal_senders, sizeof(WalSnd)));

	return fc_size;
}

/* 分配并初始化与 walsender 相关的共享内存 */
void WalSndShmemInit(void)
{
	bool		fc_found;
	int			fc_i;

	WalSndCtl = (WalSndCtlData *)
		ShmemInitStruct("Wal Sender Ctl", WalSndShmemSize(), &fc_found);

	if (!fc_found)
	{
		/* 第一次通过，所以初始化 */
		MemSet(WalSndCtl, 0, WalSndShmemSize());

		for (fc_i = 0; fc_i < NUM_SYNC_REP_WAIT_MODE; fc_i++)
			SHMQueueInit(&(WalSndCtl->SyncRepQueue[fc_i]));

		for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
		{
			WalSnd	   *fc_walsnd = &WalSndCtl->walsnds[fc_i];

			SpinLockInit(&fc_walsnd->mutex);
		}
	}
}

/*
 * 唤醒所有 walsenders
 *
 * 这将在关键部分内调用，因此抛出错误并不可取。
 */
void WalSndWakeup(void)
{
	int			fc_i;

	for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
	{
		Latch	   *fc_latch;
		WalSnd	   *fc_walsnd = &WalSndCtl->walsnds[fc_i];

		/*
		 * 获取锁存器指针，持有自旋锁，针对指针读取不是原子的不太可能情况（因为它们是8字节的）。
		 */
		SpinLockAcquire(&fc_walsnd->mutex);
		fc_latch = fc_walsnd->latch;
		SpinLockRelease(&fc_walsnd->mutex);

		if (fc_latch != NULL)
			SetLatch(fc_latch);
	}
}

/*
 * 等待 FeBe 套接字准备就绪，或超时。掩码应由可选的 WL_SOCKET_WRITEABLE 和 WL_SOCKET_READABLE 标志组成。 在主进程死掉时退出。
 */
static void fc_WalSndWait(uint32 fc_socket_events, long fc_timeout, uint32 fc_wait_event)
{
	WaitEvent	fc_event;

	ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, fc_socket_events, NULL);
	if (WaitEventSetWait(FeBeWaitSet, fc_timeout, &fc_event, 1, fc_wait_event) == 1 &&
		(fc_event.events & WL_POSTMASTER_DEATH))
		proc_exit(1);
}

/*
 * 信号所有的 walsenders 进入停止状态。
 *
 * 这将触发 walsenders 进入一个无法生成更多 WAL 的状态。有关详细信息，请参见此文件的头部。
 */
void WalSndInitStopping(void)
{
	int			fc_i;

	for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
	{
		WalSnd	   *fc_walsnd = &WalSndCtl->walsnds[fc_i];
		pid_t		fc_pid;

		SpinLockAcquire(&fc_walsnd->mutex);
		fc_pid = fc_walsnd->pid;
		SpinLockRelease(&fc_walsnd->mutex);

		if (fc_pid == 0)
			continue;

		SendProcSignal(fc_pid, PROCSIG_WALSND_INIT_STOPPING, InvalidBackendId);
	}
}

/*
 * 等待所有 WAL 发送者退出或到达停止状态。这
 * 被检查点用于控制何时可以安全地执行关机检查点。
 */
void WalSndWaitStopping(void)
{
	for (;;)
	{
		int			fc_i;
		bool		fc_all_stopped = true;

		for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
		{
			WalSnd	   *fc_walsnd = &WalSndCtl->walsnds[fc_i];

			SpinLockAcquire(&fc_walsnd->mutex);

			if (fc_walsnd->pid == 0)
			{
				SpinLockRelease(&fc_walsnd->mutex);
				continue;
			}

			if (fc_walsnd->state != WALSNDSTATE_STOPPING)
			{
				fc_all_stopped = false;
				SpinLockRelease(&fc_walsnd->mutex);
				break;
			}
			SpinLockRelease(&fc_walsnd->mutex);
		}

		/* 如果所有 WAL 发送者的确认完成，则安全离开 */
		if (fc_all_stopped)
			return;

		pg_usleep(10000L);		/* 等待 10 毫秒 */
	}
}

/* 为当前 walsender 设置状态（仅在 walsender 中调用） */
void WalSndSetState(WalSndState fc_state)
{
	WalSnd	   *fc_walsnd = MyWalSnd;

	Assert(am_walsender);

	if (fc_walsnd->state == fc_state)
		return;

	SpinLockAcquire(&fc_walsnd->mutex);
	fc_walsnd->state = fc_state;
	SpinLockRelease(&fc_walsnd->mutex);
}

/*
 * 返回一个表示状态的字符串常量。这在
 * 系统视图中使用，不应翻译。
 */
static const char * fc_WalSndGetStateString(WalSndState fc_state)
{
	switch (fc_state)
	{
		case WALSNDSTATE_STARTUP:
			return "startup";
		case WALSNDSTATE_BACKUP:
			return "backup";
		case WALSNDSTATE_CATCHUP:
			return "catchup";
		case WALSNDSTATE_STREAMING:
			return "streaming";
		case WALSNDSTATE_STOPPING:
			return "stopping";
	}
	return "UNKNOWN";
}

static Interval *
fc_offset_to_interval(TimeOffset fc_offset)
{
	Interval   *fc_result = palloc(sizeof(Interval));

	fc_result->month = 0;
	fc_result->day = 0;
	fc_result->time = fc_offset;

	return fc_result;
}

/*
 * 返回 walsenders 的活动，包括发送给备用服务器的 PID 和 xlog 位置。
 */
Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_WAL_SENDERS_COLS	12
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	SyncRepStandbyData *fc_sync_standbys;
	int			fc_num_standbys;
	int			fc_i;

	InitMaterializedSRF(fcinfo, 0);

	/*
	 * 获取当前活动的同步备用。 在我们完成之前，这可能
	 * 已经过时，但我们仍然会使用这些数据。
	 */
	fc_num_standbys = SyncRepGetCandidateStandbys(&fc_sync_standbys);

	for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
	{
		WalSnd	   *fc_walsnd = &WalSndCtl->walsnds[fc_i];
		XLogRecPtr	fc_sentPtr;
		XLogRecPtr	fc_write;
		XLogRecPtr	fc_flush;
		XLogRecPtr	fc_apply;
		TimeOffset	fc_writeLag;
		TimeOffset	fc_flushLag;
		TimeOffset	fc_applyLag;
		int			fc_priority;
		int			fc_pid;
		WalSndState fc_state;
		TimestampTz fc_replyTime;
		bool		fc_is_sync_standby;
		Datum		fc_values[PG_STAT_GET_WAL_SENDERS_COLS];
		bool		fc_nulls[PG_STAT_GET_WAL_SENDERS_COLS];
		int			fc_j;

		/* 从共享内存收集数据 */
		SpinLockAcquire(&fc_walsnd->mutex);
		if (fc_walsnd->pid == 0)
		{
			SpinLockRelease(&fc_walsnd->mutex);
			continue;
		}
		fc_pid = fc_walsnd->pid;
		fc_sentPtr = fc_walsnd->sentPtr;
		fc_state = fc_walsnd->state;
		fc_write = fc_walsnd->write;
		fc_flush = fc_walsnd->flush;
		fc_apply = fc_walsnd->apply;
		fc_writeLag = fc_walsnd->writeLag;
		fc_flushLag = fc_walsnd->flushLag;
		fc_applyLag = fc_walsnd->applyLag;
		fc_priority = fc_walsnd->sync_standby_priority;
		fc_replyTime = fc_walsnd->replyTime;
		SpinLockRelease(&fc_walsnd->mutex);

		/*
		 * 检测 walsender 是否被视为同步。 我们可以
		 * 通过同时检查 PID 和 walsnd_index 来提供一些保护
		 * 以防过时的数据。
		 */
		fc_is_sync_standby = false;
		for (fc_j = 0; fc_j < fc_num_standbys; fc_j++)
		{
			if (fc_sync_standbys[fc_j].walsnd_index == fc_i &&
				fc_sync_standbys[fc_j].pid == fc_pid)
			{
				fc_is_sync_standby = true;
				break;
			}
		}

		memset(fc_nulls, 0, sizeof(fc_nulls));
		fc_values[0] = Int32GetDatum(fc_pid);

		if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))
		{
			/*
			 * 仅超级用户和具有 pg_read_all_stats 权限的角色
			 * 可以查看详细信息。其他用户仅获取 PID 值，以了解
			 * 它是一个 walsender，但没有详细信息。
			 */
			MemSet(&fc_nulls[1], true, PG_STAT_GET_WAL_SENDERS_COLS - 1);
		}
		else
		{
			fc_values[1] = CStringGetTextDatum(fc_WalSndGetStateString(fc_state));

			if (XLogRecPtrIsInvalid(fc_sentPtr))
				fc_nulls[2] = true;
			fc_values[2] = LSNGetDatum(fc_sentPtr);

			if (XLogRecPtrIsInvalid(fc_write))
				fc_nulls[3] = true;
			fc_values[3] = LSNGetDatum(fc_write);

			if (XLogRecPtrIsInvalid(fc_flush))
				fc_nulls[4] = true;
			fc_values[4] = LSNGetDatum(fc_flush);

			if (XLogRecPtrIsInvalid(fc_apply))
				fc_nulls[5] = true;
			fc_values[5] = LSNGetDatum(fc_apply);

			/*
			 * 将备用视为 pg_basebackup 背景进程
			 * 始终返回无效的刷新位置，作为
			 * 异步备用。
			 */
			fc_priority = XLogRecPtrIsInvalid(fc_flush) ? 0 : fc_priority;

			if (fc_writeLag < 0)
				fc_nulls[6] = true;
			else
				fc_values[6] = IntervalPGetDatum(fc_offset_to_interval(fc_writeLag));

			if (fc_flushLag < 0)
				fc_nulls[7] = true;
			else
				fc_values[7] = IntervalPGetDatum(fc_offset_to_interval(fc_flushLag));

			if (fc_applyLag < 0)
				fc_nulls[8] = true;
			else
				fc_values[8] = IntervalPGetDatum(fc_offset_to_interval(fc_applyLag));

			fc_values[9] = Int32GetDatum(fc_priority);

			/*
			 * 更容易理解的备用状态版本。这纯粹是
			 * 信息性的。
			 *
			 * 在基于法定人数的同步复制中，
			 * 列在 synchronous_standby_names 中的每个备用的角色
			 * 可以非常频繁地变化。任何在一瞬间被视为"同步"的备用，
			 * 在下一刻可能被切换为"潜在"的备用。因此，
			 * 报告它们的同步状态为"同步"或"潜在"基本上是无用的。
			 * 我们仅报告"法定人数"给它们。
			 */
			if (fc_priority == 0)
				fc_values[10] = CStringGetTextDatum("async");
			else if (fc_is_sync_standby)
				fc_values[10] = SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY ?
					CStringGetTextDatum("sync") : CStringGetTextDatum("quorum");
			else
				fc_values[10] = CStringGetTextDatum("potential");

			if (fc_replyTime == 0)
				fc_nulls[11] = true;
			else
				fc_values[11] = TimestampTzGetDatum(fc_replyTime);
		}

		tuplestore_putvalues(fc_rsinfo->setResult, fc_rsinfo->setDesc,
							 fc_values, fc_nulls);
	}

	return (Datum) 0;
}

/*
 * 向备用发送保持活动消息。
 *
 * 如果 requestReply 被设置，则该消息请求另一方向我们发送
 * 返回消息，以进行心跳目的。我们还设置一个标志以
 * 让附近的代码知道我们在等待该响应，以避免
 * 重复请求。
 *
 * writePtr 是 WAL 被发送到的位置。它本质上
 * 与 sentPtr 相同，但在某些情况下，我们需要在
 * sentPtr 更新之前发送保持活动消息，例如在跳过空事务时。
 */
static void fc_WalSndKeepalive(bool fc_requestReply, XLogRecPtr fc_writePtr)
{
	elog(DEBUG2, "sending replication keepalive");

	/* 构造消息... */
	resetStringInfo(&output_message);
	pq_sendbyte(&output_message, 'k');
	pq_sendint64(&output_message, XLogRecPtrIsInvalid(fc_writePtr) ? sentPtr : fc_writePtr);
	pq_sendint64(&output_message, GetCurrentTimestamp());
	pq_sendbyte(&output_message, fc_requestReply ? 1 : 0);

	/* ...并将其包装在 CopyData 中发送 */
	pq_putmessage_noblock('d', output_message.data, output_message.len);

	/* 设置本地标志 */
	if (fc_requestReply)
		waiting_for_ping_response = true;
}

/*
 * 如果经过的时间太长，则发送保持活动消息。
 */
static void fc_WalSndKeepaliveIfNecessary(void)
{
	TimestampTz fc_ping_time;

	/*
	 * 如果全局禁用了超时，或者我们正在做一些不参与超时的事情，
	 * 则不要发送保持活动消息。
	 */
	if (wal_sender_timeout <= 0 || last_reply_timestamp <= 0)
		return;

	if (waiting_for_ping_response)
		return;

	/*
	 * 如果在没有收到任何来自备用的回复的情况下
	 * 已经过去了 wal_sender_timeout 的一半时间，则向备用发送
	 * 一个请求立即回复的保持活动消息。
	 */
	fc_ping_time = TimestampTzPlusMilliseconds(last_reply_timestamp,
											wal_sender_timeout / 2);
	if (last_processing >= fc_ping_time)
	{
		fc_WalSndKeepalive(true, InvalidXLogRecPtr);

		/* 尝试将待处理输出刷新到客户端 */
		if (pq_flush_if_writable() != 0)
			fc_WalSndShutdown();
	}
}

/*
 * 记录 WAL 的结束和它在本地刷新的时间，以便
 * LagTrackerRead 可以计算经过的时间（滞后），当这个 WAL 位置
 * 最终报告为已被写入、刷新并在回复消息中被备用应用时。
 */
static void fc_LagTrackerWrite(XLogRecPtr fc_lsn, TimestampTz fc_local_flush_time)
{
	bool		fc_buffer_full;
	int			fc_new_write_head;
	int			fc_i;

	if (!am_walsender)
		return;

	/*
	 * 如果自上次以来 lsn 没有推进，则不执行任何操作。 这样
	 * 我们仅在写入新 WAL 时记录新样本。
	 */
	if (lag_tracker->last_lsn == fc_lsn)
		return;
	lag_tracker->last_lsn = fc_lsn;

	/*
	 * 如果推进循环缓冲区的写入头会与任何
	 * 读取头相撞，那么缓冲区已满。 换句话说，最慢的读取者
	 * （推测是应用）控制空间的释放。
	 */
	fc_new_write_head = (lag_tracker->write_head + 1) % LAG_TRACKER_BUFFER_SIZE;
	fc_buffer_full = false;
	for (fc_i = 0; fc_i < NUM_SYNC_REP_WAIT_MODE; ++fc_i)
	{
		if (fc_new_write_head == lag_tracker->read_heads[fc_i])
			fc_buffer_full = true;
	}

	/*
	 * 如果缓冲区已满，目前我们只需回退一个槽并覆盖
	 * 最后一个样本，作为降低采样率的简单（尽管有些不均匀）方法。
	 * 可能有更好的自适应压缩算法。
	 */
	if (fc_buffer_full)
	{
		fc_new_write_head = lag_tracker->write_head;
		if (lag_tracker->write_head > 0)
			lag_tracker->write_head--;
		else
			lag_tracker->write_head = LAG_TRACKER_BUFFER_SIZE - 1;
	}

	/* 在当前写入头位置存储一个样本。 */
	lag_tracker->buffer[lag_tracker->write_head].lsn = fc_lsn;
	lag_tracker->buffer[lag_tracker->write_head].time = fc_local_flush_time;
	lag_tracker->write_head = fc_new_write_head;
}


/*
 * 找出 WAL 位置 'lsn'（或已知的最高早期 LSN）被本地刷新和时间 'now' 之间经过了多少时间。
 * 我们为从备用接收到的每个报告的 LSN 位置都有一个单独的读头； 'head' 控制使用哪个读头。
 * 每当读头跨越一个使用 LagTrackerWrite 写入滞后缓冲区的 LSN 时，我们可以使用相关的时间戳来
 * 查找此 LSN（或更早的 LSN）被本地刷新时的时间，因此可以计算滞后。
 *
 * 如果没有新样本数据可用，则返回 -1；否则返回经过的时间（微秒）。
 */
static TimeOffset
fc_LagTrackerRead(int fc_head, XLogRecPtr fc_lsn, TimestampTz fc_now)
{
	TimestampTz fc_time = 0;

	/* 读取所有未读样本，直到此 LSN 或缓冲区末尾。 */
	while (lag_tracker->read_heads[fc_head] != lag_tracker->write_head &&
		   lag_tracker->buffer[lag_tracker->read_heads[fc_head]].lsn <= fc_lsn)
	{
		fc_time = lag_tracker->buffer[lag_tracker->read_heads[fc_head]].time;
		lag_tracker->last_read[fc_head] =
			lag_tracker->buffer[lag_tracker->read_heads[fc_head]];
		lag_tracker->read_heads[fc_head] =
			(lag_tracker->read_heads[fc_head] + 1) % LAG_TRACKER_BUFFER_SIZE;
	}

	/*
	 * 如果滞后跟踪器为空，这意味着备用已经处理了我们发送的所有内容，因此我们现在应该清除 'last_read'。
	 * 如果我们不这样做，我们将面临在一段空闲时间后在下一个 WAL 涌流开始时使用过时和不相关的样本进行插值的风险。
	 */
	if (lag_tracker->read_heads[fc_head] == lag_tracker->write_head)
		lag_tracker->last_read[fc_head].time = 0;

	if (fc_time > fc_now)
	{
		/* 如果时钟以某种方式向后走，则视为未找到。 */
		return -1;
	}
	else if (fc_time == 0)
	{
		/*
		 * 我们没有跨越一个时间。如果还有一个未来的样本
		 * 我们尚未到达，并且我们已经到达至少一个样本，
		 * 让我们插值局部刷新时间。这主要有助于
		 * 报告一个完全停滞的应用位置，使其显示出
		 * 增加的延迟，因为否则我们必须等待它
		 * 最终再次开始移动并跨越我们的一个样本，然后
		 * 我们才能显示延迟增加。
		 */
		if (lag_tracker->read_heads[fc_head] == lag_tracker->write_head)
		{
			/* 没有未来样本，因此我们无法插值。 */
			return -1;
		}
		else if (lag_tracker->last_read[fc_head].time != 0)
		{
			/* 我们可以在 last_read 和下一个样本之间插值。 */
			double		fc_fraction;
			WalTimeSample fc_prev = lag_tracker->last_read[fc_head];
			WalTimeSample fc_next = lag_tracker->buffer[lag_tracker->read_heads[fc_head]];

			if (fc_lsn < fc_prev.lsn)
			{
				/*
				 * 报告的 LSN 通常不应该向后变化，但在
				 * 时间线变化时是可能的。视为未找到。
				 */
				return -1;
			}

			Assert(fc_prev.lsn < fc_next.lsn);

			if (fc_prev.time > fc_next.time)
			{
				/* 如果时钟以某种方式向后走，则视为未找到。 */
				return -1;
			}

			/* 查看我们在前一个和下一个样本之间有多远。 */
			fc_fraction =
				(double) (fc_lsn - fc_prev.lsn) / (double) (fc_next.lsn - fc_prev.lsn);

			/* 成比例地缩放局部刷新时间。 */
			fc_time = (TimestampTz)
				((double) fc_prev.time + (fc_next.time - fc_prev.time) * fc_fraction);
		}
		else
		{
			/*
			 * 我们只有一个未来样本，这意味着我们之前完全
			 * 跟上了，但现在有一个新的 WAL 峰值，备用服务器
			 * 尚未处理第一个样本。在备用服务器到达
			 * 未来样本之前，我们所能做的就是报告
			 * 如果现在播放该样本的假设延迟。
			 */
			fc_time = lag_tracker->buffer[lag_tracker->read_heads[fc_head]].time;
		}
	}

	/* 返回自局部刷新时间以来经过的时间（以微秒为单位）。 */
	Assert(fc_time != 0);
	return fc_now - fc_time;
}


#ifdef FDD
static bool fc_FDB_Wal_SndKeepalive(void)
{
	//int rc;
	long		fc_sleeptime = 0;
	int			fc_wakeEvents = WL_TIMEOUT;

	//检查中断
	CHECK_FOR_INTERRUPTS();

	/* 处理最近收到的任何请求或信号 */
	if (ConfigReloadPending)
	{
		ConfigReloadPending = false;
		ProcessConfigFile(PGC_SIGHUP);
		SyncRepInitConfig(MyReplicationSlot != NULL ? MyReplicationSlot->data.failover : false);
	}

	if (got_STOPPING)
		fc_WalSndShutdown();

	//处理订阅端发送的消息
	fc_ProcessRepliesIfAny();
	
	//处理是否备端已经超时
	fc_WalSndCheckTimeOut();

	//发送保活消息s
	fc_WalSndKeepalive(false,InvalidXLogRecPtr);

	waiting_for_ping_response = true;

	if (pq_flush_if_writable() != 0)
		fc_WalSndShutdown();


	fc_sleeptime = fc_WalSndComputeSleeptime(GetCurrentTimestamp());
		
	//处理中断,等待 
	fc_wakeEvents = WL_LATCH_SET | WL_EXIT_ON_PM_DEATH |
			WL_SOCKET_READABLE | WL_TIMEOUT;

	if (pq_is_send_pending())
		fc_wakeEvents |= WL_SOCKET_WRITEABLE;

	(void) WaitLatchOrSocket(MyLatch, fc_wakeEvents,
						 	MyProcPort->sock, fc_sleeptime,
						 	WAIT_EVENT_WAL_SENDER_WAIT_WAL);
	
	return true;
}
#endif

